Java8中的Adder 和 Accumulator

有这样一个需求，需要统计某个Runnable的子类在多线程环境下被执行的次数。那么代码很简单:

    public void testAdder() throws InterruptedException {

        AtomicLong atomicLong = new AtomicLong();

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 0; i < 10; i++) {
            executorService.execute(new AtomicLongTask(atomicLong));
        }

        Thread.sleep(1000);

        long l = atomicLong.get();
    }

    class AtomicLongTask implements Runnable {

        private AtomicLong atomicLong;

        public AtomicLongTask(AtomicLong atomicLong) {
            this.atomicLong = atomicLong;
        }

        @Override
        public void run() {
            atomicLong.incrementAndGet();
        }
    }

使用AtomicLong就可以解决这个问题。那么AtomicLong是怎样保证多线程功能正常的呢？经典的CPU-线程模型图来了。

image.png

每个CPU都有多个核心，假设每个核心运行一个线程，那么每个线程就会加载自己所有需要的数据到local cache中。如果使用普通的long类型的话，那么就会产生线程同步问题，这个在最开始的文章就提到过了。AtomicLong做的就是每次读之前就从shared cache读取值到local cache中，更新完之后再写回（flush）shared cache里，接着再更新（refresh）其他线程cache中的值。每个线程都使用这样的操作，就必须做好同步访问处理（AtomicLong帮我们实现好了这一步）。这样当然没有什么错，只是每次这样flush、refresh并且做同步处理确实很耗性能。有没有什么更好的解决方案呢？其实退一步想，每个线程单独更新值，在最后获取的时候把所有线程更新的值取出来，处理一下不就可以了吗。以这个
atomicLong.incrementAndGet();简单的例子来讲，就是把所有线程更新的值都取出来，再相加即可（仅针对这个简单的例子）。

这里就可以引出LongAdder来代替上面的AtomicLong，代码如下:

    public void testAdder() throws InterruptedException {
        LongAdder longAdder = new LongAdder();

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 0; i < 10; i++) {
            executorService.execute(new LongAdderTask(longAdder));
        }

        Thread.sleep(1000);

        long l1 = longAdder.sum();//对比atomicLong.get()
    }

    class LongAdderTask implements Runnable {

        private LongAdder longAdder;

        public LongAdderTask(LongAdder adder) {
            longAdder = adder;
        }

        @Override
        public void run() {
            longAdder.increment();
        }
    }

原理就是每个线程只更新local cache的变量，最后计算的时候根据所有线程的变量来获取最后的值。

image.png
image.png
这样就可以避免AtomicLong多线程同步带来的性能问题。

Accumulator是一个更具扩展性的Adder。直接上代码：

    public void testAdder() throws InterruptedException {
        LongAccumulator longAccumulator = new LongAccumulator((left, right) -> left + right, 0);

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 0; i < 10; i++) {
            executorService.execute(new LongAccumulatorTask(longAccumulator));
        }

        Thread.sleep(1000);
        

        long l = longAccumulator.get();
    }

    class LongAccumulatorTask implements Runnable{

        private LongAccumulator longAccumulator;

        public LongAccumulatorTask(LongAccumulator accumulator) {
            longAccumulator = accumulator;
        }

        @Override
        public void run() {
            longAccumulator.accumulate(1);
        }
    }

构造函数LongAccumulator((left, right) -> left + right, 0);其中0是初始值，初始化的是left，y就是下面longAccumulator.accumulate(1);传递的1，这里的功能和上面的LongAdder相同，如果需要自定义相关功能，只需要改写表达式即可。需要注意的是这个表达式里面不要出现对外部对象引用，因为这个表达式的执行顺序和线程都不确定，容易出现问题。