Java并发编程一基本类型、数组类型原子类和累加器初使用加源码分析

我不是女神ヾ 2023-07-06 11:35 18阅读 0赞

推荐：[Java并发编程汇总][Java]

## Java并发编程一基本类型、数组类型原子类和累加器初使用加源码分析 ##

首先我们来看一看有哪些原子类。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]  
现在我们来看看该如何去使用这些基本类型、数组类型原子类和累加器吧。

**AtomicInteger**

代码：

package atomic;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class AtomicIntegerDemo1 implements Runnable { 
    
        private static final AtomicInteger atomicInteger = new AtomicInteger();
    
        public void incrementAtomic() { 
            atomicInteger.incrementAndGet();
        }
    
        private static int basicCount = 0;
    
        public void incrementBasic() { 
            basicCount++;
        }
    
        public static void main(String[] args) throws InterruptedException { 
            AtomicIntegerDemo1 r = new AtomicIntegerDemo1();
            Thread t1 = new Thread(r);
            Thread t2 = new Thread(r);
            t1.start();
            t2.start();
            t1.join();
            t2.join();
            System.out.println("原子类的结果：" + atomicInteger.get());
            System.out.println("普通变量的结果：" + basicCount);
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < 10000; i++) { 
                incrementAtomic();
                incrementBasic();
            }
        }
    }

输出：

原子类的结果：20000
    普通变量的结果：18159

很明显可以看出，普通变量的结果是不正确的，出现了线程安全问题，我们一般的方法是用锁锁住`basicCount++`这一部分，而`AtomicInteger`实例的结果是正确的，为什么它是线程安全的呢？  
进入到`AtomicInteger`类的源码中，可以发现一个关键的工具`Unsafe`类。

什么是`Unsafe`类？

*  Java不能直接访问操作系统底层，而是通过本地方法来访问。`Unsafe`类提供了硬件级别的原子操作。
 *  `Unsafe`类在`sun.misc`包下，不属于Java标准。很多Java的基础类库，包括一些被广泛使用的高性能开发库都是基于`Unsafe`类开发，比如`Netty`、`Hadoop`、`Kafka`等。 `Unsafe`类是用于在实质上扩展 Java语言表达能力、便于在更高层（Java 层）代码里实现原本要在更低层（C 层）实现的核心库功能用的。
 *  这些功能包括裸内存的申请/释放/访问，低层硬件的 `atomic`/`volatile` 支持，创建未初始化对象等。
 *  它原本的设计就只应该被标准库使用，因此不建议在生产环境中使用。

// setup to use Unsafe.compareAndSwapInt for updates
        private static final Unsafe unsafe = Unsafe.getUnsafe();

`unsafe`通过反射拿到了`AtomicInteger`类的`value`属性，代码如下。

static { 
            try { 
                valueOffset = unsafe.objectFieldOffset
                    (AtomicInteger.class.getDeclaredField("value"));
            } catch (Exception ex) {  throw new Error(ex); }
        }

`AtomicInteger`类的`value`属性是用`volatile`关键字修饰的，保证了`value`属性的可见性。

private volatile int value;

`Unsafe`类通过`CAS`来保证线程安全，为什么`CAS`能保证线程安全呢？

*  `CAS`是英文单词`Compare And Swap`的缩写，翻译过来就是比较并替换。
 *  `CAS`机制当中使用了`3`个基本操作数：内存地址`V`，旧的预期值`A`，要修改的新值`B`。
 *  更新一个变量的时候，只有当变量的预期值`A`和内存地址`V`当中的实际值相同时，才会将内存地址`V`对应的值修改为`B`。

是不是有一种乐观锁的感觉，并且整个比较并替换的操作是一个原子操作，这由操作系统来支持（瞬间底气十足）。  
`Unsafe`类的`native`方法，便是基于`CAS`实现的。

public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
    
        public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
    
        public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);

而`Unsafe`类中一些常用方法是基于上面的`native`方法，也就是基于`CAS`实现的，这些方法在原子类中经常会被使用到。

public final int getAndAddInt(Object var1, long var2, int var4) { 
            int var5;
            do { 
                var5 = this.getIntVolatile(var1, var2);
            } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
    
            return var5;
        }
    
        public final long getAndAddLong(Object var1, long var2, long var4) { 
            long var6;
            do { 
                var6 = this.getLongVolatile(var1, var2);
            } while(!this.compareAndSwapLong(var1, var2, var6, var6 + var4));
    
            return var6;
        }
    
        public final int getAndSetInt(Object var1, long var2, int var4) { 
            int var5;
            do { 
                var5 = this.getIntVolatile(var1, var2);
            } while(!this.compareAndSwapInt(var1, var2, var5, var4));
    
            return var5;
        }
    
        public final long getAndSetLong(Object var1, long var2, long var4) { 
            long var6;
            do { 
                var6 = this.getLongVolatile(var1, var2);
            } while(!this.compareAndSwapLong(var1, var2, var6, var4));
    
            return var6;
        }
    
        public final Object getAndSetObject(Object var1, long var2, Object var4) { 
            Object var5;
            do { 
                var5 = this.getObjectVolatile(var1, var2);
            } while(!this.compareAndSwapObject(var1, var2, var5, var4));
    
            return var5;
        }

回到主题，我们再来看看`AtomicInteger`类还有哪些比较常用的方法。

/** * Atomically sets to the given value and returns the old value. * * @param newValue the new value * @return the previous value */
        public final int getAndSet(int newValue) { 
            return unsafe.getAndSetInt(this, valueOffset, newValue);
        }

`getAndSet(int newValue)`会设置新值并且返回旧值`Atomically sets to the given value and returns the old value`，这些都是原子的操作。

/** * Atomically increments by one the current value. * * @return the previous value */
        public final int getAndIncrement() { 
            return unsafe.getAndAddInt(this, valueOffset, 1);
        }

`getAndIncrement()`先获取值再将值加`1`。

/** * Atomically decrements by one the current value. * * @return the previous value */
        public final int getAndDecrement() { 
            return unsafe.getAndAddInt(this, valueOffset, -1);
        }

`getAndDecrement()`先获取值再将值减`1`。

/** * Atomically adds the given value to the current value. * * @param delta the value to add * @return the previous value */
        public final int getAndAdd(int delta) { 
            return unsafe.getAndAddInt(this, valueOffset, delta);
        }

`getAndAdd(int delta)`先获取值再将值加`delta`。

/** * Atomically increments by one the current value. * * @return the updated value */
        public final int incrementAndGet() { 
            return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
        }

`incrementAndGet()`先将值加`1`再获取这个新值。

/** * Atomically decrements by one the current value. * * @return the updated value */
        public final int decrementAndGet() { 
            return unsafe.get

`decrementAndGet()`先将值减`1`再获取这个新值。

/** * Atomically adds the given value to the current value. * * @param delta the value to add * @return the updated value */
        public final int addAndGet(int delta) { 
            return unsafe.getAndAddInt(this, valueOffset, delta) + delta;
        }

`addAndGet(int delta)`先将值加`delta`再获取这个新值。

就介绍这些方法，可见这些方法都用到了`Unsafe`类中的方法。

**AtomicLong**

`AtomicLong`类和`AtomicInteger`类是类似的，只不过对应的基本类型不一样而已，用法差不多，就简单介绍一下。

代码：

package atomic;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicLong;
    
    public class AtomicLongDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
            AtomicLong counter = new AtomicLong(0);
            ExecutorService service = Executors.newFixedThreadPool(20);
            long start = System.currentTimeMillis();
            for (int i = 0; i < 10000; i++) { 
                service.submit(new Task(counter));
            }
            service.shutdown();
            while (!service.isTerminated()) { 
    
            }
            long end = System.currentTimeMillis();
            System.out.println(counter.get());
            System.out.println("AtomicLong耗时：" + (end - start));
        }
    
        private static class Task implements Runnable { 
    
            private AtomicLong counter;
    
            public Task(AtomicLong counter) { 
                this.counter = counter;
            }
    
            @Override
            public void run() { 
                for (int i = 0; i < 10000; i++) { 
                    counter.incrementAndGet();
                }
            }
        }
    }

输出：

100000000
    AtomicLong耗时：1982

为什么要输出耗时呢？是为了方便和后面的其他原子类进行性能对比，先埋一个伏笔。  
`AtomicLong`类和`AtomicInteger`类的一些常用方法也都差不多，只不过`AtomicLong`类中使用`long`来进行值的操作。

**AtomicBoolean**

`AtomicBoolean`类的源码是比较短的，大家可以看看，有了前面的基础相信大家可以很容易看懂。

package java.util.concurrent.atomic;
    import sun.misc.Unsafe;
    
    public class AtomicBoolean implements java.io.Serializable { 
        private static final long serialVersionUID = 4654671469794556979L;
        // setup to use Unsafe.compareAndSwapInt for updates
        private static final Unsafe unsafe = Unsafe.getUnsafe();
        private static final long valueOffset;
    
        static { 
            try { 
                valueOffset = unsafe.objectFieldOffset
                    (AtomicBoolean.class.getDeclaredField("value"));
            } catch (Exception ex) {  throw new Error(ex); }
        }
    
        private volatile int value;
    
        /** * Creates a new {@code AtomicBoolean} with the given initial value. * * @param initialValue the initial value */
        public AtomicBoolean(boolean initialValue) { 
            value = initialValue ? 1 : 0;
        }
    
        /** * Creates a new {@code AtomicBoolean} with initial value {@code false}. */
        public AtomicBoolean() { 
        }
    
        /** * Returns the current value. * * @return the current value */
        public final boolean get() { 
            return value != 0;
        }
    
        /** * Atomically sets the value to the given updated value * if the current value {@code ==} the expected value. * * @param expect the expected value * @param update the new value * @return {@code true} if successful. False return indicates that * the actual value was not equal to the expected value. */
        public final boolean compareAndSet(boolean expect, boolean update) { 
            int e = expect ? 1 : 0;
            int u = update ? 1 : 0;
            return unsafe.compareAndSwapInt(this, valueOffset, e, u);
        }
    
        /** * Atomically sets the value to the given updated value * if the current value {@code ==} the expected value. * * <p><a href="package-summary.html#weakCompareAndSet">May fail * spuriously and does not provide ordering guarantees</a>, so is * only rarely an appropriate alternative to {@code compareAndSet}. * * @param expect the expected value * @param update the new value * @return {@code true} if successful */
        public boolean weakCompareAndSet(boolean expect, boolean update) { 
            int e = expect ? 1 : 0;
            int u = update ? 1 : 0;
            return unsafe.compareAndSwapInt(this, valueOffset, e, u);
        }
    
        /** * Unconditionally sets to the given value. * * @param newValue the new value */
        public final void set(boolean newValue) { 
            value = newValue ? 1 : 0;
        }
    
        /** * Eventually sets to the given value. * * @param newValue the new value * @since 1.6 */
        public final void lazySet(boolean newValue) { 
            int v = newValue ? 1 : 0;
            unsafe.putOrderedInt(this, valueOffset, v);
        }
    
        /** * Atomically sets to the given value and returns the previous value. * * @param newValue the new value * @return the previous value */
        public final boolean getAndSet(boolean newValue) { 
            boolean prev;
            do { 
                prev = get();
            } while (!compareAndSet(prev, newValue));
            return prev;
        }
    
        /** * Returns the String representation of the current value. * @return the String representation of the current value */
        public String toString() { 
            return Boolean.toString(get());
        }
    
    }

**AtomicArray**

原子数组类，这里就介绍一种原子数组类`AtomicIntegerArray`，其他可以类比。

代码：

package atomic;
    
    import java.util.concurrent.atomic.AtomicIntegerArray;
    
    
    public class AtomicArrayDemo { 
    
        public static void main(String[] args) { 
            AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(1000);
            Incrementer incrementer = new Incrementer(atomicIntegerArray);
            Decrementer decrementer = new Decrementer(atomicIntegerArray);
            Thread[] threadsIncrementer = new Thread[100];
            Thread[] threadsDecrementer = new Thread[100];
            for (int i = 0; i < 100; i++) { 
                threadsDecrementer[i] = new Thread(decrementer);
                threadsIncrementer[i] = new Thread(incrementer);
                threadsDecrementer[i].start();
                threadsIncrementer[i].start();
            }
    
            for (int i = 0; i < 100; i++) { 
                try { 
                    threadsDecrementer[i].join();
                    threadsIncrementer[i].join();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
            for (int i = 0; i < atomicIntegerArray.length(); i++) { 
                if (atomicIntegerArray.get(i)!=0) { 
                    System.out.println("发现了错误"+i);
                }
                System.out.println(atomicIntegerArray.get(i));
            }
            System.out.println("运行结束");
        }
    }
    
    class Decrementer implements Runnable { 
    
        private AtomicIntegerArray array;
    
        public Decrementer(AtomicIntegerArray array) { 
            this.array = array;
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < array.length(); i++) { 
                array.getAndDecrement(i);
            }
        }
    }
    
    class Incrementer implements Runnable { 
    
        private AtomicIntegerArray array;
    
        public Incrementer(AtomicIntegerArray array) { 
            this.array = array;
        }
    
        @Override
        public void run() { 
            for (int i = 0; i < array.length(); i++) { 
                array.getAndIncrement(i);
            }
        }
    }

输出：

0
    0
    0
    0
    0
    0
    0
    0
    0
    0
    0
    运行结束

我只粘贴了一部分输出，输出结果都是`0`，这也符合我们的预期。

原子数组类只不过这个数组的元素都是原子类罢了，能保证线程安全也就很容易理解了。

用法有一点不同，如下代码，操作一个元素需要指定下标，毕竟是数组嘛。

/** * Atomically adds the given value to the element at index {@code i}. * * @param i the index * @param delta the value to add * @return the previous value */
        public final int getAndAdd(int i, int delta) { 
            return unsafe.getAndAddInt(array, checkedByteOffset(i), delta);
        }

**LongAdder**

这里用`LongAdder`类来和`AtomicLong`类进行对比。

代码：

package atomic;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.LongAdder;
    
    public class LongAdderDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
            LongAdder counter = new LongAdder();
            ExecutorService service = Executors.newFixedThreadPool(20);
            long start = System.currentTimeMillis();
            for (int i = 0; i < 10000; i++) { 
                service.submit(new Task(counter));
            }
            service.shutdown();
            while (!service.isTerminated()) { 
    
            }
            long end = System.currentTimeMillis();
            System.out.println(counter.sum());
            System.out.println("LongAdder耗时：" + (end - start));
        }
    
        private static class Task implements Runnable { 
    
            private LongAdder counter;
    
            public Task(LongAdder counter) { 
                this.counter = counter;
            }
    
            @Override
            public void run() { 
                for (int i = 0; i < 10000; i++) { 
                    counter.increment();
                }
            }
        }
    }

输出：

100000000
    LongAdder耗时：144

`AtomicLong耗时：1982`与`LongAdder耗时：144`这个差距就很明显了。

因为，`LongAdder`在`AtomicLong`的基础上将单点（一个`value`）的更新压力分散到各个节点（`Cell[]`数组），在低并发的时候通过对`base`的直接更新可以很好的保障和`AtomicLong`的性能基本保持一致，而在高并发的时候通过压力分散提高了性能，缺点是`LongAdder`在统计的时候如果有并发更新，可能导致统计的数据有误差。

源码：

package java.util.concurrent.atomic;
    import java.io.Serializable;
    
    public class LongAdder extends Striped64 implements Serializable { 
        private static final long serialVersionUID = 7249069246863182397L;
    
        public LongAdder() { 
        }
    
        public void add(long x) { 
            Cell[] as; long b, v; int m; Cell a;
            if ((as = cells) != null || !casBase(b = base, b + x)) { 
                boolean uncontended = true;
                if (as == null || (m = as.length - 1) < 0 ||
                    (a = as[getProbe() & m]) == null ||
                    !(uncontended = a.cas(v = a.value, v + x)))
                    longAccumulate(x, null, uncontended);
            }
        }
    
        public void increment() { 
            add(1L);
        }
    
        public void decrement() { 
            add(-1L);
        }
    
        public long sum() { 
            Cell[] as = cells; Cell a;
            long sum = base;
            if (as != null) { 
                for (int i = 0; i < as.length; ++i) { 
                    if ((a = as[i]) != null)
                        sum += a.value;
                }
            }
            return sum;
        }
    
        public void reset() { 
            Cell[] as = cells; Cell a;
            base = 0L;
            if (as != null) { 
                for (int i = 0; i < as.length; ++i) { 
                    if ((a = as[i]) != null)
                        a.value = 0L;
                }
            }
        }
    
        public long sumThenReset() { 
            Cell[] as = cells; Cell a;
            long sum = base;
            base = 0L;
            if (as != null) { 
                for (int i = 0; i < as.length; ++i) { 
                    if ((a = as[i]) != null) { 
                        sum += a.value;
                        a.value = 0L;
                    }
                }
            }
            return sum;
        }
    
        public String toString() { 
            return Long.toString(sum());
        }
    
        public long longValue() { 
            return sum();
        }
    
        public int intValue() { 
            return (int)sum();
        }
    
        public float floatValue() { 
            return (float)sum();
        }
    
        public double doubleValue() { 
            return (double)sum();
        }
    
        private static class SerializationProxy implements Serializable { 
            private static final long serialVersionUID = 7249069246863182397L;
    
            private final long value;
    
            SerializationProxy(LongAdder a) { 
                value = a.sum();
            }
            
            private Object readResolve() { 
                LongAdder a = new LongAdder();
                a.base = value;
                return a;
            }
        }
    
        private Object writeReplace() { 
            return new SerializationProxy(this);
        }
    
        private void readObject(java.io.ObjectInputStream s)
            throws java.io.InvalidObjectException { 
            throw new java.io.InvalidObjectException("Proxy required");
        }
    
    }

关键在`Striped64`类中。

`Striped64`类中的关键源码如下：

/** * Padded variant of AtomicLong supporting only raw accesses plus CAS. * * JVM intrinsics note: It would be possible to use a release-only * form of CAS here, if it were provided. */
        @sun.misc.Contended static final class Cell { 
            volatile long value;
            Cell(long x) {  value = x; }
            final boolean cas(long cmp, long val) { 
                return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
            }
    
            // Unsafe mechanics
            private static final sun.misc.Unsafe UNSAFE;
            private static final long valueOffset;
            static { 
                try { 
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class<?> ak = Cell.class;
                    valueOffset = UNSAFE.objectFieldOffset
                        (ak.getDeclaredField("value"));
                } catch (Exception e) { 
                    throw new Error(e);
                }
            }
        }
    
        /** Number of CPUS, to place bound on table size */
        static final int NCPU = Runtime.getRuntime().availableProcessors();
    
        /** * Table of cells. When non-null, size is a power of 2. */
        transient volatile Cell[] cells;
    
        /** * Base value, used mainly when there is no contention, but also as * a fallback during table initialization races. Updated via CAS. */
        transient volatile long base;
    
        /** * Spinlock (locked via CAS) used when resizing and/or creating Cells. */
        transient volatile int cellsBusy;

是不是很清晰了`Cell[] cells`、`base`、`cellsBusy`。

当计数的时候，将`base`和各个`cell`元素里面的值进行叠加，从而得到总数。这里的问题是在计数的同时如果修改某个`cell`中元素，有可能导致计数的结果不准确。

`cells`初始长度是`2`，即将`value`默认是拆成`2`段，扩容的时机是`cellsBusy==0`，从字面意思理解应该是整个`cells`的元素都被线程占用了，那么就开始扩容到原来的两倍，看到这里，`LongAdder`优化的思路及一些细节基本就清楚了吧。

**LongAccumulator**

`LongAccumulator`类也是继承了`Striped64`类，所以它有和`LongAdder`类差不多的优点和缺点，不过`LongAccumulator`类可以定制化计算，当然也会有点局限，只能是关于两个`long`类型的数值进行操作。

/** * Creates a new instance using the given accumulator function * and identity element. * @param accumulatorFunction a side-effect-free function of two arguments * @param identity identity (initial value) for the accumulator function */
        public LongAccumulator(LongBinaryOperator accumulatorFunction,
                               long identity) { 
            this.function = accumulatorFunction;
            base = this.identity = identity;
        }

`LongBinaryOperator`接口是一个函数式接口。

@FunctionalInterface
    public interface LongBinaryOperator { 
    
        /** * Applies this operator to the given operands. * * @param left the first operand * @param right the second operand * @return the operator result */
        long applyAsLong(long left, long right);
    }

代码：

计算`[1, 666666)`的最大值，`Long.MIN_VALUE`为初始值（计算最大值，初始值肯定要很小）。

package atomic;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.LongAccumulator;
    import java.util.stream.LongStream;
    
    public class LongAccumulatorDemo { 
    
        public static void main(String[] args) { 
            LongAccumulator accumulator = new LongAccumulator((x, y) -> Math.max(x , y), Long.MIN_VALUE);
            ExecutorService executor = Executors.newFixedThreadPool(8);
            LongStream.range(1, 666666).forEach(i -> executor.submit(() -> accumulator.accumulate(i)));
    
            executor.shutdown();
            while (!executor.isTerminated()) { 
    
            }
            System.out.println(accumulator.getThenReset());
        }
    }

输出：

666665

很显然是符合预期的。

其他类似的原子类大家可以进行类比，这里就不涉及了。

[Java]: https://kaven.blog.csdn.net/article/details/104233588
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200218094711445.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200218095115900.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70