负载均衡算法之轮询

最近的工作事情比较少，于是就开是瞎折腾了

负载均衡

负载均衡大家一定不陌生了，一句话就是，人人有饭吃，还吃得饱，它的核心关键字就在于均衡，关于负载均衡大家基本可以脱口而出常见的几种，轮询，随机，哈希，带权值的轮询，客户端请求数等等

轮询

作为最简单的一种负载均衡策略，轮询的优点显而易见，简单，并且在多数的情况是，基本适用（一般部署的线上集群机器，大部分的配置都比较相近，差距不会那么大，因此使用轮询是一种可以接受的方案）

实现

轮询的实现简单来说就是从一个“循环列表”中不断的获取，这里的列表可以是数组，也可以是链表，也可以是map的key集合，简而言之，就是一维数组类型。

这里我简单的做了三种轮询的实现，分别是基于 Atomic包的实现，synchronized同步，以及blockingQueue

Atomic

atomic包内的类是基于cas来实现值的同步，因此可以利用这一点来做轮询，测试代码如下

	private List<Integer> list = Lists.newArrayList(1, 2, 3, 4, 5, 6);
	
    @Test
    public void test() {
        AtomicInteger atomicInteger = new AtomicInteger(0);

		// 线程数，来模拟并发的激烈程度
        int threadNum = 4;
        int total = 10_0000;

        long now = System.currentTimeMillis();
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        CountDownLatch latch = new CountDownLatch(threadNum);
        for (int i = 0; i < threadNum; i++) {
            executorService.submit(new Task(latch, atomicInteger, total));
        }
        try {
            latch.await(60L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("costs = " + (System.currentTimeMillis() - now));
    }
	
	
    private class Task implements Runnable {

        private CountDownLatch latch;
        AtomicInteger atomicInteger;
        private int total;

        Task(CountDownLatch latch, AtomicInteger atomicInteger, int total) {
            this.latch = latch;
            this.atomicInteger = atomicInteger;
            this.total = total;
        }

        @Override
        public void run() {
            long tid = Thread.currentThread().getId();
            try {
                for (int i = 0; i < this.total; i++) {
                    int idx = atomicInteger.getAndIncrement() % 6;
                    idx = list.get(idx < 0 ? -idx : idx);
//                    System.out.printf("【Thread - %d】 get=%d
", tid, idx);
                }
            } finally {
                this.latch.countDown();
            }
        }
    }

synchronized

同步是我们最容易想到的方式了

	private LinkedList<Integer> linkedList  = new LinkedList<>();
    private final Object MUTEX = new Object();
    
    @Test
    public void testSync(){
        linkedList.add(1);
        linkedList.add(2);
        linkedList.add(3);
        linkedList.add(4);
        linkedList.add(5);
        linkedList.add(6);

        long now = System.currentTimeMillis();
        int threadNum = 64;
        int total = 10_0000;

        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        CountDownLatch latch = new CountDownLatch(threadNum);
        for (int i = 0; i < threadNum; i++) {
            executorService.submit(new TaskSync(latch, total));
        }
        try {
            latch.await(120L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("costs = " + (System.currentTimeMillis() - now));

    }
    
    private class TaskSync implements Runnable{
        private CountDownLatch latch;

        private int total;

        TaskSync(CountDownLatch latch, int total) {
            this.latch = latch;
            this.total = total;
        }

        public void run() {
            long tid = Thread.currentThread().getId();
            try {
                for (int i = 0; i < this.total; i++) {
                    synchronized (MUTEX){
                        // 从头取出，并放回到队尾
                        Integer idx = linkedList.removeFirst();
//                        System.out.printf("【Thread - %d】 get=%d
", tid, idx);
                        linkedList.add(idx);
                    }
                }
            } finally {
                this.latch.countDown();
            }
        }
    }

阻塞队列

concurrent包中有很多为我们封装了底层细节的包，可以直接进行使用，其中就包含了阻塞队列，阻塞队列许多的操作都是线程安全的。

private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(16);


@Test
public void testBlocking() throws InterruptedException {
    queue.add(1);
    queue.add(2);
    queue.add(3);
    queue.add(4);
    queue.add(5);
    queue.add(6);


	long now = System.currentTimeMillis();

	int threadNum = 8;
	int total = 1000_0000;

	ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
	CountDownLatch latch = new CountDownLatch(threadNum);
	for (int i = 0; i < threadNum; i++) {
		executorService.submit(new TaskBlocking(latch, total));
	}
	try {
		latch.await(120L, TimeUnit.SECONDS);
	} catch (InterruptedException e) {
		e.printStackTrace();
	}
	System.out.println("costs = " + (System.currentTimeMillis() - now));
}


private class TaskBlocking implements Runnable {
	private CountDownLatch latch;

	private int total;

    TaskBlocking(CountDownLatch latch, int total) {
    	this.latch = latch;
    	this.total = total;
    }

    @Override
    public void run() {
        long tid = Thread.currentThread().getId();
        try {
            for (int i = 0; i < this.total; i++) {
                Integer idx = queue.poll();
                // poll可能会得到null，因此如果得到null，那么本次不算，重新获取
                if (idx == null) {
                    i--;
                    continue;
                }
                //System.out.printf("【Thread - %d】 get=%d
", tid, idx);
                queue.add(idx);
            }
        } finally {
            this.latch.countDown();
        }
    }
}

测试结果

本人机器，win10系统，CPU4核

atomic

并发数	循环次数(万次)	耗时(s)
4	10	0.08
8	10	0.12
16	10	0.135
32	10	0.16
4	1000	1.1
8	1000	2.2
16	1000	4.4
32	1000	9.5

synchronized

并发数	循环次数(万次)	耗时(s)
4	10	0.203
8	10	0.243
16	10	0.339
32	10	0.996
4	1000	3.7
8	1000	7.1
16	1000	14.4
32	1000	26.4

blocking

并发数	循环次数(万次)	耗时(s)
4	10	0.138
8	10	0.2
16	10	0.381
32	10	0.769
4	1000	4
8	1000	7.9
16	1000	20.3
32	1000	74.8

结果比较

从耗时的结果上来看，atomic是最快的一种实现，blocking最慢（blocking的取和存，在源代码中都有使用到ReentrantLock，因此一次Run()需要2次锁的获取），而synchronized会比阻塞队列的方式稍微好点

好了，以上是我对轮询的一点小探索，如果您觉得有哪里不正确或有其他建议的地方，欢迎拍砖