一 限流算法与应用

限流是对系统的一种保护措施。即限制流量请求的频率（每秒处理多少个请 求）。一般来说，当请求流量超过系统的瓶颈，则丢弃掉多余的请求流量，保 证系统的可用性。即要么不放进来，放进来的就保证提供服务。

1.1 计数器

1. 概述
计数器采用简单的计数操作，到一段时间节点后自动清零。

2. 实现

public class Counter {
    public static void main(String[] args) {
        //计数器，这里用信号量实现
        final Semaphore semaphore = new Semaphore(3);
        //定时器，到点清零
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                semaphore.release(3);
            }
        }, 3000, 3000, TimeUnit.MILLISECONDS);
        //模拟无数个请求从天而降
        while (true) {
            try {
                //判断计数器
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //如果准许响应，打印一个ok
            System.out.println("ok");
        }
    }
}

3. 结果分析
3个ok一组呈现，到下一个计数周期之前被阻断

4. 优缺点
实现起来非常简单。控制力度太过于简略，假如1s内限制3次，那么如果3次在 前100ms内已经用完，后面的900ms将只能处于阻塞状态，白白浪费掉。

5. 应用
使用计数器限流的场景较少，因为它的处理逻辑不够灵活。最常见的可能在 web的登录密码验证，输入错误次数冻结一段时间的场景。如果网站请求使用 计数器，那么恶意攻击者前100ms吃掉流量计数，使得后续正常的请求被全部 阻断，整个服务很容易被搞垮。

1.2 漏桶算法

1. 概述
漏桶算法将请求缓存在桶中，服务流程匀速处理。超出桶容量的部分丢弃。漏 桶算法主要用于保护内部的处理业务，保障其稳定有节奏的处理请求，但是无 法根据流量的波动弹性调整响应能力。现实中，类似容纳人数有限的服务大厅 开启了固定的服务窗口。

image

public class Barrel {
    public static void main(String[] args) {
//桶，用阻塞队列实现，容量为3
        final LinkedBlockingQueue<Integer> que = new LinkedBlockingQueue(3);
//定时器，相当于服务的窗口，2s处理一个
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int v = que.poll();
                System.out.println("处理：" + v);
            }
        }, 2000, 2000, TimeUnit.MILLISECONDS);
//无数个请求，i 可以理解为请求的编号
        int i = 0;
        while (true) {
            i++;
            try {
                System.out.println("put:" + i);
//如果是put，会一直等待桶中有空闲位置，不会丢弃
// que.put(i);
//等待1s如果进不了桶，就溢出丢弃
                que.offer(i, 1000, TimeUnit.MILLISECONDS);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

3. 结果分析

image

http {
    #$binary_remote_addr 表示通过remote_addr这个标识来做key，也就是限制同一客户端ip地址。
    #zone=one:10m 表示生成一个大小为10M，名字为one的内存区域，用来存储访问的频次信息。
    #rate=1r/s 表示允许相同标识的客户端每秒1次访问
    limit_req_zone $binary_remote_addr zone=one:10m rate=1r/s;
    server {
        location /limited/ {
        #zone=one 与上面limit_req_zone 里的name对应。
        #burst=5 缓冲区，超过了访问频次限制的请求可以先放到这个缓冲区内，类似代码中的队列长度。
        #nodelay 如果设置，超过访问频次而且缓冲区也满了的时候就会直接返回503，如果没有设置，则所有请求
        会等待排队，类似代码中的put还是offer。
        limit_req zone=one burst=5 nodelay;
    }
}

1.3 令牌桶

1. 概述

image

public class Token {
    public static void main(String[] args) throws InterruptedException {
//令牌桶，信号量实现，容量为3
        final Semaphore semaphore = new Semaphore(3);
//定时器，1s一个，匀速颁发令牌
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                if (semaphore.availablePermits() < 3) {
                    semaphore.release();
                }
// System.out.println("令牌数："+semaphore.availablePermits());
            }
        }, 1000, 1000, TimeUnit.MILLISECONDS);
//等待，等候令牌桶储存
        Thread.sleep(5);
//模拟洪峰5个请求，前3个迅速响应，后两个排队
        for (int i = 0; i < 5; i++) {
            semaphore.acquire();
            System.out.println("洪峰：" + i);
        }
//模拟日常请求，2s一个
        for (int i = 0; i < 3; i++) {
            Thread.sleep(1000);
            semaphore.acquire();
            System.out.println("日常：" + i);
            Thread.sleep(1000);
        }
//再次洪峰
        for (int i = 0; i < 5; i++) {
            semaphore.acquire();
            System.out.println("洪峰：" + i);
        }
//检查令牌桶的数量
        for (int i = 0; i < 5; i++) {
            Thread.sleep(2000);
            System.out.println("令牌剩余：" + semaphore.availablePermits());
        }
    }
}

3. 结果分析

image

cloud:
  gateway:
    routes:
    ‐ id: limit_route
      uri: http://localhost:8080/test
      filters:
      ‐ name: RequestRateLimiter
        args:
          #限流的key，ipKeyResolver为spring中托管的Bean，需要扩展KeyResolver接口
          key‐resolver: '#{@ipResolver}'
          #令牌桶每秒填充平均速率，相当于代码中的发放频率
          redis‐rate‐limiter.replenishRate: 1
          #令牌桶总容量，相当于代码中，信号量的容量
          redis‐rate‐limiter.burstCapacity: 3

1.4 滑动窗口

1. 概述
滑动窗口可以理解为细分之后的计数器，计数器粗暴的限定1分钟内的访问次 数，而滑动窗口限流将1分钟拆为多个段，不但要求整个1分钟内请求数小于上 限，而且要求每个片段请求数也要小于上限。相当于将原来的计数周期做了多 个片段拆分。更为精细。

image

image

package com.yungtay.mpu.util;

public class Window {
    //整个窗口的流量上限，超出会被限流
    final int totalMax = 5;
    //每片的流量上限，超出同样会被拒绝，可以设置不同的值
    final int sliceMax = 5;
    //分多少片
    final int slice = 3;
    //窗口，分3段，每段1s，也就是总长度3s
    final LinkedList<Long> linkedList = new LinkedList<>();
    //计数器，每片一个key，可以使用HashMap，这里为了控制台保持有序性和可读性，采用TreeMap
    Map<Long, AtomicInteger> map = new TreeMap();
    //心跳，每1s跳动1次，滑动窗口向前滑动一步，实际业务中可能需要手动控制滑动窗口的时机。
    ScheduledExecutorService service = Executors.newScheduledThreadPool(1);

    //获取key值，这里即是时间戳（秒）
    private Long getKey() {
        return System.currentTimeMillis() / 1000;
    }

    public Window() {
//初始化窗口，当前时间指向的是最末端，前两片其实是过去的2s
        Long key = getKey();
        for (int i = 0; i < slice; i++) {
            linkedList.addFirst(key‐i);
            map.put(key‐i, new AtomicInteger(0));
        }
//启动心跳任务，窗口根据时间，自动向前滑动，每秒1步
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                Long key = getKey();
//队尾添加最新的片
                linkedList.addLast(key);
                map.put(key, new AtomicInteger());
//将最老的片移除
                map.remove(linkedList.getFirst());
                linkedList.removeFirst();
                System.out.println("step:" + key + ":" + map);
                ;
            }
        }, 1000, 1000, TimeUnit.MILLISECONDS);
    }

    //检查当前时间所在的片是否达到上限
    public boolean checkCurrentSlice() {
        long key = getKey();
        AtomicInteger integer = map.get(key);
        if (integer != null) {
            return integer.get() < sliceMax;
        }
//默认允许访问
        return true;
    }

    //检查整个窗口所有片的计数之和是否达到上限
    public boolean checkAllCount() {
        return map.values().stream().mapToInt(value ‐ > value.get()).sum() < totalMax;
    }

    //请求来临....
    public void req() {
        Long key = getKey();
//如果时间窗口未到达当前时间片，稍微等待一下
//其实是一个保护措施，放置心跳对滑动窗口的推动滞后于当前请求
        while (linkedList.getLast() < key) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
//开始检查，如果未达到上限，返回ok，计数器增加1
//如果任意一项达到上限，拒绝请求，达到限流的目的
//这里是直接拒绝。现实中可能会设置缓冲池，将请求放入缓冲队列暂存
        if (checkCurrentSlice() && checkAllCount()) {
            map.get(key).incrementAndGet();
            System.out.println(key + "=ok:" + map);
        } else {
            System.out.println(key + "=reject:" + map);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Window window = new Window();
//模拟10个离散的请求，相对之间有200ms间隔。会造成总数达到上限而被限流
        for (int i = 0; i < 10; i++) {
            Thread.sleep(200);
            window.req();
        }
//等待一下窗口滑动，让各个片的计数器都置零
        Thread.sleep(3000);
//模拟突发请求，单个片的计数器达到上限而被限流
        System.out.println("‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐");
        for (int i = 0; i < 10; i++) {
            window.req();
        }
    }
}

3. 结果分析

image

二 定时算法与应用

系统或者项目中难免会遇到各种需要自动去执行的任务，实现这些任务的手段 也多种多样，如操作系统的crontab，spring框架的quartz，java的Timer和 ScheduledThreadPool都是定时任务中的典型手段。

2.1 最小堆

1. 概述
Timer是java中最典型的基于优先级队列+最小堆实现的定时器，内部维护一个 存放定时任务的优先级队列，该优先级队列使用了最小堆排序。当我们调用 schedule方法的时候，一个新的任务被加入queue，堆重排，始终保持堆顶是 执行时间最小（即最近马上要执行）的。同时，内部相当于起了一个线程不断 扫描队列，从队列中依次获取堆顶元素执行，任务得到调度。

2. 案例
介绍优先级队列+最小堆算法的实现原理：

class Task extends TimerTask {
    @Override
    public void run() {
        System.out.println("running...");
    }
}

public class TimerDemo {
    public static void main(String[] args) {
        Timer t = new Timer();
//在1秒后执行,以后每2秒跑一次
        t.schedule(new Task(), 1000, 2000);
    }
}

3. 源码分析
新加任务时，t.schedule方法会add到队列

void add(TimerTask task) {
// Grow backing store if necessary
        if (size + 1 == queue.length)
            queue = Arrays.copyOf(queue, 2 * queue.length);
        queue[++size] = task;
        fixUp(size);
    }

add实现了容量维护，不足时扩容，同时将新任务追加到队列队尾，触发堆排 序，始终保持堆顶元素最小线。

//最小堆排序
    private void fixUp(int k) {
        while (k > 1) {
//k指针指向当前新加入的节点，也就是队列的末尾节点，j为其父节点
            int j = k >> 1;
//如果新加入的执行时间比父节点晚，那不需要动
            if (queue[j].nextExecutionTime <= queue[k].nextExecutionTime)
                break;
//如果大于其父节点，父子交换
            TimerTask tmp = queue[j]; queue[j] = queue[k]; queue[k] = tmp;
//交换后，当前指针继续指向新加入的节点，继续循环，知道堆重排合格
            k = j;
        }
    }

线程调度中的run，主要调用内部mainLoop()方法，使用while循环

private void mainLoop() {
        while (true) {
            try {
                TimerTask task;
                boolean taskFired;
                synchronized(queue) {
//...
// Queue nonempty; look at first evt and do the right thing
                    long currentTime, executionTime;
                    task = queue.getMin();
                    synchronized(task.lock) {
//...
//当前时间
                        currentTime = System.currentTimeMillis();
//要执行的时间
                        executionTime = task.nextExecutionTime;
//判断是否到了执行时间
                        if (taskFired = (executionTime<=currentTime)) {
//判断下一次执行时间，单次的执行完移除
//循环的修改下次执行时间
                            if (task.period == 0) { // Non‐repeating, remove
                                queue.removeMin();
                                task.state = TimerTask.EXECUTED;
                            } else { // Repeating task, reschedule
//下次时间的计算有两种策略
//1.period是负数,那下一次的执行时间就是当前时间‐period
//2.period是正数,那下一次就是该任务本次的执行时间+period
//注意！这两种策略大不相同。因为Timer是单线程的
//如果是1，那么currentTime是当前时间，就受任务执行长短影响
//如果是2，那么executionTime是绝对时间戳，与任务长短无关
                                queue.rescheduleMin(
                                        task.period<0 ? currentTime ‐ task.period
: executionTime + task.period);
                            }
                        }
                    }
//不到执行时间，等待
                    if (!taskFired) // Task hasn't yet fired; wait
                        queue.wait(executionTime ‐ currentTime);
                }
//到达执行时间，run！
                if (taskFired) // Task fired; run it, holding no locks
                    task.run();
            } catch(InterruptedException e) {
            }
        }
    }

4. 应用

本节使用Timer为了介绍算法原理，但是Timer已过时，实际应用中推荐使用
ScheduledThreadPoolExecutor（同样内部使用DelayedWorkQueue和最小堆排序）
Timer是单线程，一旦一个失败或出现异常，将打断全部任务队列，线程池不会
Timer在jdk1.3+，而线程池需要jdk1.5+

2.2 时间轮

1. 概述
时间轮是一种更为常见的定时调度算法，各种操作系统的定时任务调度， linux crontab，基于java的通信框架Netty等。其灵感来源于我们生活中的时 钟。轮盘实际上是一个头尾相接的环状数组，数组的个数即是插槽数，每个插 槽中可以放置任务。以1天为例，将任务的执行时间%12，根据得到的数值，放 置在时间轮上，小时指针沿着轮盘扫描，扫到的点取出任务执行：

image

image

2. 实现

public class RoundTask {
    //延迟多少秒后执行
    int delay;
    //加入的序列号，只是标记一下加入的顺序
    int index;

    public RoundTask(int index, int delay) {
        this.index = index;
        this.delay = delay;
    }

    void run() {
        System.out.println("task " + index + " start , delay = " + delay);
    }

    @Override
    public String toString() {
        return String.valueOf(index + "=" + delay);
    }
}

时间轮算法：

public class RoundDemo {
    //小轮槽数
    int size1 = 10;
    //大轮槽数
    int size2 = 5;
    //小轮，数组，每个元素是一个链表
    LinkedList<RoundTask>[] t1 = new LinkedList[size1];
    //大轮
    LinkedList<RoundTask>[] t2 = new LinkedList[size2];
    //小轮计数器，指针跳动的格数，每秒加1
    final AtomicInteger flag1 = new AtomicInteger(0);
    //大轮计数器，指针跳动个格数，即每10s加1
    final AtomicInteger flag2 = new AtomicInteger(0);
    //调度器，拖动指针跳动
    ScheduledExecutorService service = Executors.newScheduledThreadPool(2);

    public RoundDemo() {
//初始化时间轮
        for (int i = 0; i < size1; i++) {
            t1[i] = new LinkedList<>();
        }
        for (int i = 0; i < size2; i++) {
            t2[i] = new LinkedList<>();
        }
    }

    //打印时间轮的结构，数组+链表
    void print() {
        System.out.println("t1:");
        for (int i = 0; i < t1.length; i++) {
            System.out.println(t1[i]);
        }
        System.out.println("t2:");
        for (int i = 0; i < t2.length; i++) {
            System.out.println(t2[i]);
        }
    }

    //添加任务到时间轮
    void add(RoundTask task) {
        int delay = task.delay;
        if (delay < size1) {
//10以内的，在小轮
            t1[delay].addLast(task);
        } else {
//超过小轮的放入大轮，槽除以小轮的长度
            t2[delay / size1].addLast(task);
        }
    }

    void startT1() {
//每秒执行一次，推动时间轮旋转，取到任务立马执行
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int point = flag1.getAndIncrement() % size1;
                System.out.println("t1 ‐‐‐‐‐> slot " + point);
                LinkedList<RoundTask> list = t1[point];
                if (!list.isEmpty()) {
//如果当前槽内有任务，取出来，依次执行，执行完移除
                    while (list.size() != 0) {
                        list.getFirst().run();
                        list.removeFirst();
                    }
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    void startT2() {
//每10秒执行一次，推动时间轮旋转，取到任务下方到t1
        service.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int point = flag2.getAndIncrement() % size2;
                System.out.println("t2 =====> slot " + point);
                LinkedList<RoundTask> list = t2[point];
                if (!list.isEmpty()) {
//如果当前槽内有任务，取出，放到定义的小轮
                    while (list.size() != 0) {
                        RoundTask task = list.getFirst();
//放入小轮哪个槽呢？小轮的槽按10取余数
                        t1[task.delay % size1].addLast(task);
//从大轮中移除
                        list.removeFirst();
                    }
                }
            }
        }, 0, 10, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
        RoundDemo roundDemo = new RoundDemo();
//生成100个任务，每个任务的延迟时间随机
        for (int i = 0; i < 100; i++) {
            roundDemo.add(new RoundTask(i, new Random().nextInt(50)));
        }
        //打印，查看时间轮任务布局
        roundDemo.print();
//启动大轮
        roundDemo.startT2();
//小轮启动
        roundDemo.startT1();
    }
}

3. 结果分析

   
   
   image

三 负载均衡算法

负载均衡，英文名称为Load Balance，其含义就是指将负载（工作任务）进行 平衡、分摊到多个操作单元上进行运行，例如FTP服务器、Web服务器、企业核 心应用服务器和其它主要任务服务器等，从而协同完成工作任务。既然涉及到 多个机器，就涉及到任务如何分发，这就是负载均衡算法问题。

3.1 轮询（RoundRobin）

1. 概述
轮询即排好队，一个接一个。前面调度算法中用到的时间片轮转，就是一种典 型的轮询。但是前面使用数组和下标轮询实现。这里尝试手动写一个双向链表 形式实现服务器列表的请求轮询算法。

2. 实现

public class RR {
    class Server {
        Server prev;
        Server next;
        String name;

        public Server(String name) {
            this.name = name;
        }
    }

    //当前服务节点
    Server current;

    //初始化轮询类，多个服务器ip用逗号隔开
    public RR(String serverName) {
        System.out.println("init server list : " + serverName);
        String[] names = serverName.split(",");
        for (int i = 0; i < names.length; i++) {
            Server server = new Server(names[i]);
            if (current == null) {
//如果当前服务器为空，说明是第一台机器，current就指向新创建的server
                this.current = server;
//同时，server的前后均指向自己。
                current.prev = current;
                current.next = current;
            } else {
//否则说明已经有机器了，按新加处理。
                addServer(names[i]);
            }
        }
    }

    //添加机器
    void addServer(String serverName) {
        System.out.println("add server : " + serverName);
        Server server = new Server(serverName);
        Server next = this.current.next;
//在当前节点后插入新节点
        this.current.next = server;
        server.prev = this.current;
//修改下一节点的prev指针
        server.next = next;
        next.prev = server;
    }

    //将当前服务器移除，同时修改前后节点的指针，让其直接关联
//移除的current会被回收器回收掉
    void remove() {
        System.out.println("remove current = " + current.name);
        this.current.prev.next = this.current.next;
        this.current.next.prev = this.current.prev;
        this.current = current.next;
    }

    //请求。由当前节点处理即可
//注意：处理完成后，current指针后移
    void request() {
        System.out.println(this.current.name);
        this.current = current.next;
    }

    public static void main(String[] args) throws InterruptedException {
//初始化两台机器
        RR rr = new RR("192.168.0.1,192.168.0.2");
//启动一个额外线程，模拟不停的请求
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    rr.request();
                }
            }
        }).start();
//3s后，3号机器加入清单
        Thread.currentThread().sleep(3000);
        rr.addServer("192.168.0.3");
//3s后，当前服务节点被移除
        Thread.currentThread().sleep(3000);
        rr.remove();
    }
}

3. 结果分析

image

3.2 随机（Random）

1. 概述
从可服务的列表中随机取一个提供响应。随机存取的场景下，适合使用数组更 高效的实现下标随机读取。

2. 实现
定义一个数组，在数组长度内取随机数，作为其下标即可。非常简单

public class Rand {
    ArrayList<String> ips;

    public Rand(String nodeNames) {
        System.out.println("init list : " + nodeNames);
        String[] nodes = nodeNames.split(",");
//初始化服务器列表，长度取机器数
        ips = new ArrayList<>(nodes.length);
        for (String node : nodes) {
            ips.add(node);
        }
    }

    //请求
    void request() {
//下标，随机数，注意因子
        int i = new Random().nextInt(ips.size());
        System.out.println(ips.get(i));
    }

    //添加节点，注意，添加节点会造成内部数组扩容
//可以根据实际情况初始化时预留一定空间
    void addnode(String nodeName) {
        System.out.println("add node : " + nodeName);
        ips.add(nodeName);
    }

    //移除
    void remove(String nodeName) {
        System.out.println("remove node : " + nodeName);
        ips.remove(nodeName);
    }

    public static void main(String[] args) throws InterruptedException {
        Rand rd = new Rand("192.168.0.1,192.168.0.2");
//启动一个额外线程，模拟不停的请求
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    rd.request();
                }
            }
        }).start();
//3s后，3号机器加入清单
        Thread.currentThread().sleep(3000);
        rd.addnode("192.168.0.3");
//3s后，当前服务节点被移除
        Thread.currentThread().sleep(3000);
        rd.remove("192.168.0.2");
    }
}

3. 结果分析

image

3.1 源地址哈希（Hash）

1. 概述
对当前访问的ip地址做一个hash值，相同的key被路由到同一台机器去。场景 常见于分布式集群环境下，用户登录时的请求路由和会话保持。

2. 实现
使用HashMap可以实现请求值到对应节点的服务，其查找时的时间复杂度为 o(1)。固定一种算法，将请求映射到key上即可。举例，将请求的来源ip末 尾，按机器数取余作为key：

public class Hash {
    ArrayList<String> ips;

    public Hash(String nodeNames) {
        System.out.println("init list : " + nodeNames);
        String[] nodes = nodeNames.split(",");
//初始化服务器列表，长度取机器数
        ips = new ArrayList<>(nodes.length);
        for (String node : nodes) {
            ips.add(node);
        }
    }

    //添加节点，注意，添加节点会造成内部Hash重排，思考为什么呢？？？
//这是个问题！在一致性hash中会进入详细探讨
    void addnode(String nodeName) {
        System.out.println("add node : " + nodeName);
        ips.add(nodeName);
    }

    //移除
    void remove(String nodeName) {
        System.out.println("remove node : " + nodeName);
        ips.remove(nodeName);
    }

    //映射到key的算法，这里取余数做下标
    private int hash(String ip) {
        int last = Integer.valueOf(ip.substring(ip.lastIndexOf(".") + 1, ip.length()));
        return last % ips.size();
    }

    //请求
//注意，这里和来访ip是有关系的，采用一个参数，表示当前的来访ip
    void request(String ip) {
//下标
        int i = hash(ip);
        System.out.println(ip + "‐‐>" + ips.get(i));
    }

    public static void main(String[] args) {
        Hash hash = new Hash("192.168.0.1,192.168.0.2");
        for (int i = 1; i < 10; i++) {
//模拟请求的来源ip
            String ip = "192.168.1." + i;
            hash.request(ip);
        }
        hash.addnode("192.168.0.3");
        for (int i = 1; i < 10; i++) {
//模拟请求的来源ip
            String ip = "192.168.1." + i;
            hash.request(ip);
        }
    }
}

3. 结果分析

image

2.4 加权轮询（WRR）

1. 概述
WeightRoundRobin，轮询只是机械的旋转，加权轮询弥补了所有机器一视同仁 的缺点。在轮询的基础上，初始化时，机器携带一个比重。
2. 实现
维护一个链表，每个机器根据权重不同，占据的个数不同。轮询时权重大的，个数多，自然取到的次数变大。举个
例子：a，b，c 三台机器，权重分别为4，2，1，排位后会是a,a,a,a,b,b,c，每次请求时，从列表中依次取节点，下
次请求再取下一个。到末尾时，再从头开始。
但是这样有一个问题：机器分布不够均匀，扎堆出现了....
解决：为解决机器平滑出现的问题，nginx的源码中使用了一种平滑的加权轮询的算法，规则如下：
每个节点两个权重，weight和currentWeight，weight永远不变是配置时的值，current不停变化
变化规律如下：选择前所有current+=weight，选current最大的响应，响应后让它的current-=total
统计

image

public class WRR {
    class Node {
        int weight, currentWeight;
        String name;

        public Node(String name, int weight) {
            this.name = name;
            this.weight = weight;
            this.currentWeight = 0;
        }

        @Override
        public String toString() {
            return String.valueOf(currentWeight);
        }
    }

    //所有节点的列表
    ArrayList<Node> list;
    //总权重
    int total;

    //初始化节点列表，格式：a#4,b#2,c#1
    public WRR(String nodes) {
        String[] ns = nodes.split(",");
        list = new ArrayList<>(ns.length);
        for (String n : ns) {
            String[] n1 = n.split("#");
            int weight = Integer.valueOf(n1[1]);
            list.add(new Node(n1[0], weight));
            total += weight;
        }
    }

    //获取当前节点
    Node getCurrent() {
//执行前，current加权重
        for (Node node : list) {
            node.currentWeight += node.weight;
        }
//遍历，取权重最高的返回
        Node current = list.get(0);
        int i = 0;
        for (Node node : list) {
            if (node.currentWeight > i) {
                i = node.currentWeight;
                current = node;
            }
        }
        return current;
    }

    //响应
    void request() {
//获取当前节点
        Node node = this.getCurrent();
//第一列，执行前的current
        System.out.print(list.toString() + "‐‐‐");
//第二列，选中的节点开始响应
        System.out.print(node.name + "‐‐‐");
//响应后，current减掉total
        node.currentWeight ‐=total;
//第三列，执行后的current
        System.out.println(list);
    }

    public static void main(String[] args) {
        WRR wrr = new WRR("a#4,b#2,c#1");
//7次执行请求，看结果
        for (int i = 0; i < 7; i++) {
            wrr.request();
        }
    }
}

3. 结果分析

image

3.5 加权随机（WR）

1. 概述
WeightRandom，机器随机被筛选，但是做一组加权值，根据权值不同，选中的 概率不同。在这个概念上，可以认为随机是一种等权值的特殊情况。

2. 实现
设计思路依然相同，根据权值大小，生成不同数量的节点，节点排队后，随机获取。这里的数据结构主要涉及到随
机的读取，所以优选为数组。
与随机相同的是，同样为数组随机筛选，不同在于，随机只是每台机器1个，加权后变为多个。

public class WR {
    //所有节点的列表
    ArrayList<String> list;

    //初始化节点列表
    public WR(String nodes) {
        String[] ns = nodes.split(",");
        list = new ArrayList<>();
        for (String n : ns) {
            String[] n1 = n.split("#");
            int weight = Integer.valueOf(n1[1]);
            for (int i = 0; i < weight; i++) {
                list.add(n1[0]);
            }
        }
    }

    void request() {
//下标，随机数，注意因子
        int i = new Random().nextInt(list.size());
        System.out.println(list.get(i));
    }

    public static void main(String[] args) {
        WR wr = new WR("a#2,b#1");
        for (int i = 0; i < 9; i++) {
            wr.request();
        }
    }
}

3. 结果分析

   
   
   image

3.6 最小连接数（LC）

1. 概述
LeastConnections，即统计当前机器的连接数，选最少的去响应新的请求。前 面的算法是站在请求维度，而最小连接数是站在机器的维度。
2. 实现
定义一个链接表记录机器的节点id和机器连接数量的计数器。内部采用最小堆 做排序处理，响应时取堆顶节点即是最小连接数。

public class LC {
    //节点列表
    Node[] nodes;

    //初始化节点，创建堆
// 因为开始时各节点连接数都为0，所以直接填充数组即可
    LC(String ns) {
        String[] ns1 = ns.split(",");
        nodes = new Node[ns1.length + 1];
        for (int i = 0; i < ns1.length; i++) {
            nodes[i + 1] = new Node(ns1[i]);
        }
    }

    //节点下沉，与左右子节点比对，选里面最小的交换
//目的是始终保持最小堆的顶点元素值最小
//i:要下沉的顶点序号
    void down(int i) {
//顶点序号遍历，只要到1半即可，时间复杂度为O(log2n)
        while (i << 1 < nodes.length) {
//左子，为何左移1位？回顾一下二叉树序号
            int left = i << 1;
//右子，左+1即可
            int right = left + 1;
//标记，指向 本节点，左、右子节点里最小的，一开始取i自己
            int flag = i;
//判断左子是否小于本节点
            if (nodes[left].get() < nodes[i].get()) {
                flag = left;
            }
//判断右子
            if (right < nodes.length && nodes[flag].get() > nodes[right].get()) {
                flag = right;
            }
//两者中最小的与本节点不相等，则交换
            if (flag != i) {
                Node temp = nodes[i];
                nodes[i] = nodes[flag];
                nodes[flag] = temp;
                i = flag;
            } else {
//否则相等，堆排序完成，退出循环即可
                break;
            }
        }
    }

    //请求。非常简单，直接取最小堆的堆顶元素就是连接数最少的机器
    void request() {
        System.out.println("‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐‐");
//取堆顶元素响应请求
        Node node = nodes[1];
        System.out.println(node.name + " accept");
//连接数加1
        node.inc();
//排序前的堆
        System.out.println("before:" + Arrays.toString(nodes));
//堆顶下沉
        down(1);
//排序后的堆
        System.out.println("after:" + Arrays.toString(nodes));
    }

    public static void main(String[] args) {
//假设有7台机器
        LC lc = new LC("a,b,c,d,e,f,g");
//模拟10个请求连接
        for (int i = 0; i < 10; i++) {
            lc.request();
        }
    }

    class Node {
        //节点标识
        String name;
        //计数器
        AtomicInteger count = new AtomicInteger(0);

        public Node(String name) {
            this.name = name;
        }

        //计数器增加
        public void inc() {
            count.getAndIncrement();
        }

        //获取连接数
        public int get() {
            return count.get();
        }

        @Override
        public String toString() {
            return name + "=" + count;
        }
    }
}

3. 结果分析

image

3.7 应用案例

1. nginx upstream

upstream frontend {
  #源地址hash
  ip_hash;
  server 192.168.0.1:8081;
  server 192.168.0.2:8082 weight=1 down;
  server 192.168.0.3:8083 weight=2;
  server 192.168.0.4:8084 weight=3 backup;
  server 192.168.0.5:8085 weight=4 max_fails=3 fail_timeout=30s;
}

image

#设置负载均衡策略 eureka‐application‐service为调用的服务的名称
eureka‐applicationservice.
ribbon.NFLoadBalancerRuleClassName=com.netflix.loadbalancer.RandomRule

3. dubbo负载均衡
使用Service注解

@Service(loadbalance = "roundrobin",weight = 100)

image

四 加密算法与应用

4.1 散列

1. 概述
严格来讲这不算是一种加密，而应该叫做信息摘要算法。该算法使用散列函数 把消息或数据压缩成摘要，使得数据量变小，将数据的格式固定下来。通过数 据打乱混合，重新创建一个叫做 散列值
2. 常见算法

image

其次是文件校验场景。例如从某站下载的文件（尤其是大文件，比如系统镜像 iso），官方网站都会放置一个签名（可能是MD5，或者SHA），当用户拿到文 件后，可以本地执行散列算法与官网签名比对是否一致，来判断文件是否被篡 改。如ubuntu20.04的镜像：

image

<dependency>
  <groupId>commons‐codec</groupId>
  <artifactId>commons‐codec</artifactId>
  <version>1.14</version>
</dependency>

public class Hash {
    /**
     * jdk的security实现md5
     * 也可以借助commons‐codec包
     */
    public static String md5(String src) {
        byte[] pwd = null;
        try {
            pwd = MessageDigest.getInstance("md5").digest(src.getBytes("utf‐8"));
        } catch (Exception e) {
            e.printStackTrace();
        }
        String code = new BigInteger(1, pwd).toString(16);
        for (int i = 0; i < 32 ‐code.length();
        i++){
            code = "0" + code;
        }
        return code;
    }

    public static String commonsMd5(String src) {
        return DigestUtils.md5Hex(src);
    }

    /**
     * jdk实现sha算法
     * 也可以借助commons‐codec包
     */
    public static String sha(String src) throws Exception {
        MessageDigest sha = MessageDigest.getInstance("sha");
        byte[] shaByte = sha.digest(src.getBytes("utf‐8"));
        StringBuffer code = new StringBuffer();
        for (int i = 0; i < shaByte.length; i++) {
            int val = ((int) shaByte[i]) & 0xff;
            if (val < 16) {
                code.append("0");
            }
            code.append(Integer.toHexString(val));
        }
        return code.toString();
    }

    public static String commonsSha(String src) throws Exception {
        return DigestUtils.sha1Hex(src);
    }

    public static void main(String[] args) throws Exception {
        String name = "架构师训练营";
        System.out.println(name);
        System.out.println(md5(name));
        System.out.println(commonsMd5(name));
        System.out.println(sha(name));
        System.out.println(commonsSha(name));
    }
}

5. 结果分析

image

4.2 对称

1. 概述
加密与解密用的都是同一个秘钥，性能比非对称加密高很多。
2. 常见算法
常见的对称加密算法有 DES、3DES、AES
DES算法在POS、ATM、磁卡及智能卡（IC卡）、加油站、高速公路收费站等领域被广泛应用，以此来实现关键数
据的保密，如信用卡持卡人的PIN的加密传输，IC卡与POS间的双向认证、金融交易数据包的MAC校验等
3DES是DES加密算法的一种模式，是DES的一个更安全的变形。从DES向AES的过渡算法

image

public class AES {
    public static void main(String[] args) throws Exception {
//生成KEY
        KeyGenerator keyGenerator = KeyGenerator.getInstance("AES");
        keyGenerator.init(128);
//key转换
        Key key = new SecretKeySpec(keyGenerator.generateKey().getEncoded(), "AES");
        Cipher cipher = Cipher.getInstance("AES/ECB/PKCS5Padding");
        String src = "架构师训练营";
        System.out.println("明文：" + src);
//加密
        cipher.init(Cipher.ENCRYPT_MODE, key);
        byte[] result = cipher.doFinal(src.getBytes());
        System.out.println("加密：" + Base64.encodeBase64String(result));
//解密
        cipher.init(Cipher.DECRYPT_MODE, key);
        result = cipher.doFinal(result);
        System.out.println("解密：" + new String(result));
    }
}

5. 结果分析

image

4,3 非对称

1. 概述
非对称即加密与解密不是同一把钥匙，而是分成公钥和私钥。私钥在个人手 里，公钥公开。这一对钥匙一个用于加密，另一个用于解密。使用其中一个加 密后，则原始明文只能用对应的另一个密钥解密，即使最初用于加密的密钥
也不能用作解密。正是因为这种特性，所以称为非对称加密。

2. 常见算法
RSA、ElGamal、背包算法、Rabin（RSA的特例）、迪菲－赫尔曼密钥交换协议 中的公钥加密算法、椭圆曲线加密算法（英语：Elliptic Curve Cryptography, ECC）。使用最广泛的是RSA算法（发明者Rivest、Shmir和
Adleman姓氏首字母缩写）

image

3. 应用
最常见的，两点：https和数字签名。
严格意义上讲，https并非所有请求都使用非对称。基于性能考虑，https先使 用非对称约定一个key，后期使用该key进行对称加密和数据传输。

image

image

public class RSAUtil {
    static String privKey;
    static String publicKey;

    public static void main(String[] args) throws Exception {
//生成公钥和私钥
        genKeyPair();
//加密字符串
        String message = "架构师训练营";
        System.out.println("明文：" + message);
        System.out.println("随机公钥为:" + publicKey);
        System.out.println("随机私钥为:" + privKey);
        String messageEn = encrypt(message, publicKey);
        System.out.println("公钥加密:" + messageEn);
        String messageDe = decrypt(messageEn, privKey);
        System.out.println("私钥解密:" + messageDe);
    }

    /**
     * 随机生成密钥对
     */
    public static void genKeyPair() throws NoSuchAlgorithmException {
// KeyPairGenerator类用于生成公钥和私钥对，基于RSA算法生成对象
        KeyPairGenerator keyPairGen = KeyPairGenerator.getInstance("RSA");
// 初始化密钥对生成器，密钥大小为96‐1024位
        keyPairGen.initialize(1024, new SecureRandom());
// 生成一个密钥对，保存在keyPair中
        KeyPair keyPair = keyPairGen.generateKeyPair();
        privKey = new String(Base64.encodeBase64((keyPair.getPrivate().getEncoded())));
        publicKey = new String(Base64.encodeBase64(keyPair.getPublic().getEncoded()));
    }

    /**
     * RSA公钥加密
     */
    public static String encrypt(String str, String publicKey) throws Exception {
//base64编码的公钥
        byte[] decoded = Base64.decodeBase64(publicKey);
        RSAPublicKey pubKey = (RSAPublicKey) KeyFactory.getInstance("RSA").generatePublic(new X509EncodedKeySpec(decoded));
//RSA加密
        Cipher cipher = Cipher.getInstance("RSA");
        cipher.init(Cipher.ENCRYPT_MODE, pubKey);
        String outStr = Base64.encodeBase64String(cipher.doFinal(str.getBytes("UTF‐8")));
        return outStr;
    }

    /**
     * RSA私钥解密
     */
    public static String decrypt(String str, String privateKey) throws Exception {
//64位解码加密后的字符串
        byte[] inputByte = Base64.decodeBase64(str.getBytes("UTF‐8"));
        byte[] decoded = Base64.decodeBase64(privateKey);
        RSAPrivateKey priKey = (RSAPrivateKey) KeyFactory.getInstance("RSA").generatePrivate(new
                PKCS8EncodedKeySpec(decoded));
        Cipher cipher = Cipher.getInstance("RSA");
        cipher.init(Cipher.DECRYPT_MODE, priKey);
        return new String(cipher.doFinal(inputByte));
    }
}

5. 结果分析

image

五 一致性hash及应用

1. 背景
负载均衡策略中，我们提到过源地址hash算法，让某些请求固定的落在对应的 服务器上。这样可以解决会话信息保留的问题。

同时，标准的hash，如果机器节点数发生变更。那么请求会被重新hash，打破 了原始的设计初衷，怎么解决呢？一致性hash上场。

2. 原理

以4台机器为例，一致性hash的算法如下：
首先求出各个服务器的哈希值，并将其配置到0～232的圆上
然后采用同样的方法求出存储数据的键的哈希值，也映射圆上
从数据映射到的位置开始顺时针查找，将数据保存到找到的第一个服务器上
如果到最大值仍然找不到，就取第一个。这就是为啥形象的称之为环

image

image

image

image

public class Hash {
    //服务器列表
    private static String[] servers = { "192.168.0.1",
            "192.168.0.2", "192.168.0.3", "192.168.0.4" };
    //key表示服务器的hash值，value表示服务器
    private static SortedMap<Integer, String> serverMap = new TreeMap<Integer, String>();
    static {
        for (int i=0; i<servers.length; i++) {
            int hash = getHash(servers[i]);
//理论上，hash环的最大值为2^32
//这里为做实例，将ip末尾作为上限也就是254
//那么服务器是0‐4，乘以60后可以均匀分布到 0‐254 的环上去
//实际的请求ip到来时，在环上查找即可
            hash *= 60;
            System.out.println("add " + servers[i] + ", hash=" + hash);
            serverMap.put(hash, servers[i]);
        }
    }
    //查找节点
    private static String getServer(String key) {
        int hash = getHash(key);
//得到大于该Hash值的所有server
        SortedMap<Integer, String> subMap = serverMap.tailMap(hash);
        if(subMap.isEmpty()){
//如果没有比该key的hash值大的，则从第一个node开始
            Integer i = serverMap.firstKey();
//返回对应的服务器
            return serverMap.get(i);
        }else{
//第一个Key就是顺时针过去离node最近的那个结点
            Integer i = subMap.firstKey();
//返回对应的服务器
            return subMap.get(i);
        }
    }
    //运算hash值
//该函数可以自由定义，只要做到取值离散即可
//这里取ip地址的最后一节
    private static int getHash(String str) {
        String last = str.substring(str.lastIndexOf(".")+1,str.length());String last = str.substring(str.lastIndexOf(".")+1,str.length());
        return Integer.valueOf(last);
    }
    public static void main(String[] args) {
//模拟5个随机ip请求
        for (int i = 1; i < 8; i++) {
            String ip = "192.168.1."+ i*30;
            System.out.println(ip +" ‐‐‐> "+getServer(ip));
        }
//将5号服务器加到2‐3之间，取中间位置，150
        System.out.println("add 192.168.0.5，hash=150");
        serverMap.put(150,"192.168.0.5");
//再次发起5个请求
        for (int i = 1; i < 8; i++) {
            String ip = "192.168.1."+ i*30;
            System.out.println(ip +" ‐‐‐> "+getServer(ip));
        }
    }
}

5. 验证

image

六 典型业务场景应用

6.1 网站敏感词过滤

1. 场景
敏感词、文字过滤是一个网站必不可少的功能，高效的过滤算法是非常有必要 的。针对过滤首先想到的可能是这样：
2. 方案
方案一、使用java里的String contains，逐个遍历敏感词：

String[] s = "广告,广告词,中奖".split(",");
String text = "讨厌的广告词";
boolean flag = false;
for (String s1 : s) {
  if (text.contains(s1)){
    flag = true;
    break;
  }
}
System.out.println(flag);

方案二、正则表达式：

System.out.println(text.matches(".*(广告|广告词|中奖).*"));

其实无论采取哪个方法，基本是换汤不换药。都是整体字符匹配，效率值得商 榷。那怎么办呢？DFA算法出场。
3. 概述
DFA即Deterministic Finite Automaton，也就是确定有穷自动机，它是是通 过event和当前的state得到下一个state，即event+state=nextstate。

对照到以上案例，查找和停止查找是动作，找没找到是状态，每一步的查找和 结果决定下一步要不要继续。DFA算法在敏感词上应用的关键是构建敏感词 库，如果我们把以上案例翻译成json表达如下：

{
    "isEnd": 0,
    "广": {
    "isEnd": 0,
    "告": {
    "isEnd": 1,
    "词": {
    "isEnd": 1
    }
    }
    },
    "中": {
    "isEnd": 0,
    "奖": {
    "isEnd": 1
    }
    }
}

查找过程如下：首先把text按字拆分，逐个字查找词库的key，先从“讨”开 始，没有就下一个字“厌”，直到“广”，找到就判断isEnd，如果为1，说明匹配 成功包含敏感词，如果为0，那就继续匹配“告”，直到isEnd=1为止。

匹配策略上，有两种。最小和最大匹配。最小则匹配【广告】，最大则需要匹 配到底【广告词】

4. java实现
先加入fastjson坐标，查看敏感词库结构要用到

<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>fastjson</artifactId>
  <version>1.2.70</version>
</dependency>

/**
 * 敏感词处理DFA算法
 */
public class SensitiveWordUtil {
    //短匹配规则，如：敏感词库["广告","广告词"]，语句："我是广告词"，匹配结果：我是[广告]
    public static final int SHORT_MATCH = 1;
    //长匹配规则，如：敏感词库["广告","广告词"]，语句："我是广告词"，匹配结果：我是[广告词]
    public static final int LONG_MATCH = 2;
    /**
     * 敏感词库
     */
    public static HashMap sensitiveWordMap;

    /**
     * 初始化敏感词库
     * words:敏感词，多个用英文逗号分隔
     */
    private static void initSensitiveWordMap(String words) {
        String[] w = words.split(",");
        sensitiveWordMap = new HashMap(w.length);
        Map nowMap;
        for (String key : w) {
            nowMap = sensitiveWordMap;
            for (int i = 0; i < key.length(); i++) {
//转换成char型
                char keyChar = key.charAt(i);
//库中获取关键字
                Map wordMap = (Map) nowMap.get(keyChar);
//如果不存在新建一个，并加入词库
                if (wordMap == null) {
                    wordMap = new HashMap();
                    wordMap.put("isEnd", "0");
                    nowMap.put(keyChar, wordMap);
                }
                nowMap = wordMap;
                if (i == key.length() ‐1){
//最后一个
                    nowMap.put("isEnd", "1");
                }
            }
        }
    }

    /**
     * 判断文字是否包含敏感字符
     *
     * @return 若包含返回true，否则返回false
     */
    public static boolean contains(String txt, int matchType) {
        for (int i = 0; i < txt.length(); i++) {
            int matchFlag = checkSensitiveWord(txt, i, matchType); //判断是否包含敏感字符
            if (matchFlag > 0) { //大于0存在，返回true
                return true;
            }
        }
        return false;
    }

    /**
     * 沿着文本字符挨个往后检索文字中的敏感词
     */
    public static Set<String> getSensitiveWord(String txt, int matchType) {
        Set<String> sensitiveWordList = new HashSet<>();
        for (int i = 0; i < txt.length(); i++) {
//判断是否包含敏感字符
            int length = checkSensitiveWord(txt, i, matchType);
            if (length > 0) {//存在,加入list中
                sensitiveWordList.add(txt.substring(i, i + length));
//指针沿着文本往后移动敏感词的长度
//也就是一旦找到敏感词，加到列表后，越过这个词的字符，继续往下搜索
//但是必须减1，因为for循环会自增，如果不减会造成下次循环跳格而忽略字符
//这会造成严重误差
                i = i + length ‐1;
            }
//如果找不到，i就老老实实一个字一个字的往后移动，作为begin进行下一轮
        }
        return sensitiveWordList;
    }

    /**
     * 从第beginIndex个字符的位置，往后查找敏感词
     * 如果找到，返回敏感词字符的长度，不存在返回0
     * 这个长度用于找到后提取敏感词和后移指针，是个性能关注点
     */
    private static int checkSensitiveWord(String txt, int beginIndex, int matchType) {
//敏感词结束标识位：用于敏感词只有1位的情况
        boolean flag = false;
//匹配到的敏感字的个数，也就是敏感词长度
        int length = 0;
        char word;
//从根Map开始查找
        Map nowMap = sensitiveWordMap;
        for (int i = beginIndex; i < txt.length(); i++) {
//被判断语句的第i个字符开始
            word = txt.charAt(i);
            //获取指定key，并且将敏感库指针指向下级map
            nowMap = (Map) nowMap.get(word);
            if (nowMap != null) {//存在，则判断是否为最后一个
//找到相应key，匹配长度+1
                length++;
//如果为最后一个匹配规则,结束循环，返回匹配标识数
                if ("1".equals(nowMap.get("isEnd"))) {
//结束标志位为true
                    flag = true;
//短匹配，直接返回,长匹配还需继续查找
                    if (SHORT_MATCH == matchType) {
                        break;
                    }
                }
            } else {
//敏感库不存在，直接中断
                break;
            }
        }
        if (length < 2 || !flag) {
//长度必须大于等于1才算是词，字的话就不必这么折腾了
            length = 0;
        }
        return length;
    }

    public static void main(String[] args) {
//初始化敏感词库
        SensitiveWordUtil.initSensitiveWordMap("广告,广告词,中奖");
        System.out.println("敏感词库结构：" + JSON.toJSONString(sensitiveWordMap));
        String string = "关于中奖广告的广告词筛选";
        System.out.println("被检测文本：" + string);
        System.out.println("待检测字数：" + string.length());
//是否含有关键字
        boolean result = SensitiveWordUtil.contains(string, SensitiveWordUtil.LONG_MATCH);
        System.out.println("长匹配：" + result);
        result = SensitiveWordUtil.contains(string, SensitiveWordUtil.SHORT_MATCH);
        System.out.println("短匹配：" + result);
//获取语句中的敏感词
        Set<String> set =
                SensitiveWordUtil.getSensitiveWord(string, SensitiveWordUtil.LONG_MATCH);
        System.out.println("长匹配到：" + set);
        set = SensitiveWordUtil.getSensitiveWord(string, SensitiveWordUtil.SHORT_MATCH);
        System.out.println("短匹配到：" + set);
    }
}

6.2 最优商品topk

1. 背景
topk是一个典型的业务场景，除了最优商品，包括推荐排名、积分排名所有涉 及到排名前k的地方都是该算法的应用场合。

topk即得到一个集合后，筛选里面排名前k个数值。问题看似简单，但是里面 的数据结构和算法体现着对解决方案性能的思索和深度挖掘。到底有几种方 法，这些方案里蕴含的优化思路究竟是怎么样的？这节来讨论

2. 方案

image

public class Topk {
    //堆元素下沉，形成最小堆，序号从i开始
    static void down(int[] nodes, int i) {
//顶点序号遍历，只要到1半即可，时间复杂度为O(log2n)
        while (i << 1 < nodes.length) {
//左子，为何左移1位？回顾一下二叉树序号
            int left = i << 1;
//右子，左+1即可
            int right = left + 1;
//标记，指向 本节点，左、右子节点里最小的，一开始取i自己
            int flag = i;
//判断左子是否小于本节点
            if (nodes[left] < nodes[i]) {
                flag = left;
            }
//判断右子
            if (right < nodes.length && nodes[flag] > nodes[right]) {
                flag = right;
            }
//两者中最小的与本节点不相等，则交换
            if (flag != i) {
                int temp = nodes[i];
                nodes[i] = nodes[flag];
                nodes[flag] = temp;
                i = flag;
            } else {
//否则相等，堆排序完成，退出循环即可
                break;
            }
        }
    }

    public static void main(String[] args) {
//原始数据
        int[] src = {3, 6, 2, 7, 4, 8, 1, 9, 2, 5};
//要取几个
        int k = 5;
//堆，为啥是k+1？请注意，最小堆的0是无用的，序号从1开始
        int[] nodes = new int[k + 1];
//取前k个数，注意这里只是个二叉树，还不满足最小堆的要求
        for (int i = 0; i < k; i++) {
            nodes[i + 1] = src[i];
        }
        System.out.println("before:" + Arrays.toString(nodes));
//从最底的子树开始，堆顶下沉
//这里才真正的形成最小堆
        for (int i = k >> 1; i >= 1; i‐‐){
            down(nodes, i);
        }
        System.out.println("create:" + Arrays.toString(nodes));
//把余下的n‐k个数，放到堆顶，依次下沉，topk堆算法的开始
        for (int i = src.length ‐k;
        i<src.length ;
        i++){
            if (nodes[1] < src[i]) {
                nodes[1] = src[i];
                down(nodes, 1);
            }
        }
        System.out.println("topk:" + Arrays.toString(nodes));
    }
}

4. 结果分析

   
   
   image

相关文章

基于数据结构和算法的业务应用一