java 原子类累加器

港控/mmm° 2022-12-21 04:56 197阅读 0赞

## 前言 ##

本文所有代码请添加vm参数运行`-XX:-RestrictContended`防止注解失效。

1.  `LongAccumulator`
2.  `DoubleAccumulator`

在多线程进程情况，我们可能希望计算某一共享变量，如页面访问次数，商品剩余数量秒杀。本文仅用`LongAccumulator`做说明。正确的使用累加器可以提供并发效率，如下面的测试案例中两个都是多线程安全的无锁方式，但是效率相差14倍。

## `LongAccumulator` ##

### 基础使用 ###

public static void main(String[] args) throws InterruptedException {
        
    
            LongAccumulator longAccumulator = new LongAccumulator(new LongBinaryOperator() {
        
                @Override
                public long applyAsLong(long left, long right) {
        
                    //当前值和最新需要累加值
                    //left指向当前longAccumulator存储的值，right是需要累加的值
                    //比如longAccumulator.accumulate(2); 那么right为2
                    return left + right;
                }
            }, 10);
    
            //加2
            longAccumulator.accumulate(2);
    
            //输出12
            System.out.println(longAccumulator.get());
    
            //重置为初始化的数值为10
            longAccumulator.reset();
    
            //输出10
            System.out.println(longAccumulator.get());
    
            //累加当前值，运行后为7
            longAccumulator.accumulate(-3);
            //得到当前值然后重置为初始值10
            System.out.println(longAccumulator.getThenReset());
    
    
        }

其实上述代码可以用`AtomicLong`实现，但是为什么`JUC`却而外提供了这样一个类供我们使用。

我们构造30个线程，每个线程堆共享变量自增`10000000`次。

public static void main(String[] args) throws InterruptedException {
        
    
            long start = System.currentTimeMillis();
            LongAccumulator longAccumulator = 
            new LongAccumulator(((left, right) -> left + right), 0);
            
            Runnable run = () -> {
        
                int i = 10000000;
                while (--i > 0) {
        
                    longAccumulator.accumulate(1);
                }
            };
    
            //构造1000个线程
            List<Thread> collect = IntStream
            .range(0, 30)
            .mapToObj((int value) -> new Thread(run))
            .collect(Collectors.toList());
            
            //启动他们
            collect.forEach(Thread::start);
            //等候他们完成
            collect.forEach((thread -> {
        
                try {
        
                    thread.join();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }));
    
            System.out.println("结果" + longAccumulator.get());
            long end = System.currentTimeMillis();
            System.out.println("耗时" + (end - start));
    
        }

输出：

结果299999970
    耗时439

我们在用`AtomicLong`

public static void main(String[] args) throws InterruptedException {
        
    
            long start = System.currentTimeMillis();
            AtomicLong longAccumulator = new AtomicLong();
            Runnable run = () -> {
        
                int i = 10000000;
                while (--i > 0) {
        
                    longAccumulator.incrementAndGet();
                }
            };
            //构造1000个线程
            List<Thread> collect = IntStream
            .range(0, 30)
            .mapToObj((int value) -> new Thread(run))
            .collect(Collectors.toList());
            
            //启动他们
            collect.forEach(Thread::start);
            //等候他们完成
            collect.forEach((thread -> {
        
                try {
        
                    thread.join();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }));
    
            System.out.println("结果" + longAccumulator.get());
            long end = System.currentTimeMillis();
            System.out.println("耗时" + (end - start));
    
        }

输出：

结果299999970
    耗时6240

为了正确理解效率相差之大我们必须学习原理

### CPU缓存原理 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center]

上图中每个`Core1 Core2 ...` 表示每个`cpu`核心。  
由于`cpu`计算速度远远大于内存读写速度，所以我们往往会在内存和缓存之间加入若干级别的缓存硬件，越靠近cpu的我们编号越低 速度也越快，比如`L1`缓存读写速度大于`L2`，而每个CPU核心都有自己独立的缓存器，这也带来了可见性相关问题。（可见性问题这里不在过多说明）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 1]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 2]  
`L1`缓存器和内存最小的交换的单位是缓存行，即如果内存想要缓存`变量A`,会同时缓存内存的其他变量直到缓存行的大小。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 3]  
上图缓存器中存在多个缓存行

**缓存内存变量流程:**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 4]

缓存器会一次性缓存内存中一个缓存行大小的。

**Cpu修改缓存变量**  
假设`Cpu`修改`L1缓存`中的`变量A`,那么`变量A`所在缓存行的所有变量会全部更新到内存中，而不是单纯`变量A`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 5]

**Cpu缓存更新带来的问题**

缓存虽然提高了整体的计算能力，但是也带来部分弊端。  
假设我们有多个`Cpu1,Cpu2...` 需要同时对`共享变量A`进行读写操作，而对`共享变量B`仅进行读取操作，那么很有可能`变量A,变量B`被放入同一个缓存行中，由于`变量A`需要时常同步到内存和缓存中，导致读取`变量B`时也要等候缓存完成才能读取。

如下图所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 7]

**那么我们有什么办法可以解决呢？**  
`空间换时间`方式可以让我们避免`变量A`和`变量B`被放入一个同一个缓存行。我们在变量A临近的附近插入无关变量直到填充整个缓存行大小。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 8]

**案例代码：**

案例来自参考文章修改

class PersonValue {
        
        volatile public long value;
    }
    
    
    class Person extends Thread {
        
    
        public final static long ITERATIONS = 500L * 1000L * 1000L;
        static PersonValue[] personValuesArr = new PersonValue[2];
    
        static {
        
            personValuesArr[0] = new PersonValue();
            personValuesArr[1] = new PersonValue();
        }
    
        int index = 0;
    
        public Person(int index) {
        
            this.index = index;
        }
    
        @Override
        public void run() {
        
    
            long i = ITERATIONS;
            while (--i > 0) {
        
                personValuesArr[index].value = i;
    
            }
        }
    }
    
    public class TestContend {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            long start = System.currentTimeMillis();
    
            Person person = new Person(0);
            Person person1 = new Person(1);
    
    
            person.start();
            person1.start();
    
            person.join();
            person1.join();
    
            long end = System.currentTimeMillis();
            System.out.println("耗时" + TimeUnit.MILLISECONDS.toSeconds(end - start));
        }
    }

我们看下输出的耗时：  
`耗时22`

我们简单修改`PersonValue`类

class PersonValue {
        
        //填充缓存行
        volatile long p0, p1, p2, p3, p4, p5, p6;
        volatile public long value;
        //填充缓存行
        volatile long q0, q1, q2, q3, q4, q5, q6;
    }

输出结果：  
`耗时3`

为了方便我们这样做JDK提供了注解,下面注解可以省去我们手动写填充代码。

@Contended
    class PersonValue {
        
        volatile public long value;
    }

运行这个注解请添加vm参数`-XX:-RestrictContended`

另一种形式的代码也是同理：

@Contended
    class PersonValue {
        
        volatile public long value;
    }
    
    
    class Person extends Thread {
        
    
        public final static long ITERATIONS = 500L * 1000L * 1000L*4;
    
        //    static PersonValue[] personValuesArr = new PersonValue[2];
        public static PersonValue personValuesA = new PersonValue();
        public static PersonValue personValuesB = new PersonValue();
    
    //    static {
        
    //        personValuesArr[0] = new PersonValue();
    //        personValuesArr[1] = new PersonValue();
    //    }
    
        int index = 0;
    
        public Person(int index) {
        
            this.index = index;
        }
    
        @Override
        public void run() {
        
    
            long i = ITERATIONS;
            while (--i > 0) {
        
                if (index == 0) {
        
                    personValuesA.value = i;
                } else {
        
                    personValuesB.value = i;
                }
    //            personValuesArr[index].value = i;
    
            }
        }
    }

但需要注意以下代码是不符合填充的:  
静态共享变量不能使用上述伪代码

class Person extends Thread {
        
    
        public final static long ITERATIONS = 500L * 1000L * 1000L*20;
    
        public static int personValuesA = 3;
        @Contended
        public static int personValuesB = 1;
    
        int index = 0;
    
        public Person(int index) {
        
            this.index = index;
        }
    
        @Override
        public void run() {
        
            long i = ITERATIONS;
            while (--i > 0) {
        
                if (index == 0) {
        
                    personValuesA = 3;
                } else {
        
                    personValuesB = 3;
                }
            }
        }
    }

输出：

耗时3

我们去掉标记后

class Person extends Thread {
        
    
    	//...
        public static int personValuesA = 3;
        public static int personValuesB = 1;
        //...
    }

输出：

耗时3

你可以将内部属性抽离成一个类：

class PersonValue {
        
        volatile public int valueA;
        @Contended
        volatile public int valueB;
    }
    
    class Person extends Thread {
        
    
        public final static long ITERATIONS = 500L * 1000L * 1000L ;
        public static volatile PersonValue personValue = new PersonValue();
        int index = 0;
        public Person(int index) {
        
            this.index = index;
        }
    
        @Override
        public void run() {
        
            long i = ITERATIONS;
            while (--i > 0) {
        
                if (index == 0) {
        
                    personValue.valueA = 3;
                } else {
        
                    personValue.valueB = 3;
                }
            }
        }
    }

关于为什么静态字段不能使用上述代码避免伪共享原因，笔者也没有清晰的答案。只能猜测`java`静态属性只有读取才会载入内存,且只会加载被读取属性而不是整个类。当然这是我的个人意淫想法，我提问在`Stack Overflow`了坐等大神解决。

[Java false sharing in multiple thread problem][]  
笔者意淫图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 9]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 10]

### 源码分析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 11]

理解原理后笔者看了一遍源码，笔者居然在concurrentHashmap分析过，这里就不写了。  
[ConcurrentHashMap addCount分析][ConcurrentHashMap addCount]

## 参考 ##

[Java8使用@sun.misc.Contended避免伪共享][Java8_sun.misc.Contended]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/7f403dbe6b70407798cc7f709f52627d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/50902c912f404256a48fe51e96ce606f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/c150a3d521f0465886166d1ec95bd323.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/10e10575bbf543f8a0a5bc45e81b0a6b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/738e2daf895c4432ab9fd52ce85f03b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/9545b0dd3bc94098ac893c3c58b3f4b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/48c563686da847e5a3606599803e107c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221120/f5eec6781352414385c18a4e07837ff3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221120/f8e419197a144f6e85b2fecae497301b.png
[Java false sharing in multiple thread problem]: https://stackoverflow.com/questions/64845841/java-false-sharing-in-multiple-thread-problem
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221120/2f455ccdfe514c8790c2e6ae6ab163b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221120/f221442e22d64be19bb6bf1f32f6d49b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FmYW5taW5neWlx_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221120/edb5bd54d54149a29215ac5199dddca2.png
[ConcurrentHashMap addCount]: https://fanmingyi.blog.csdn.net/article/details/108950356
[Java8_sun.misc.Contended]: https://www.jianshu.com/p/c3c108c3dcfd