Java高并发（四） - Java 原子类详解

Java高并发（一）- 并发编程的几个基本概念
 Java高并发（二） - Java 内存模型与线程
 Java高并发（三） - CountDownLatch、CyclicBarrier和Semaphore
Java高并发（四） - Java 原子类详解
 Java高并发（五） - 线程安全策略
 Java高并发（六） - 锁的优化及 JVM 对锁优化所做的努力

CountDownLatch，Semaphore 基础请查看之前的文章 CountDownLatch、CyclicBarrier 和 Semaphore

线程不安全的高并发实现

客户端模拟执行 5000 个任务，线程数量是 200，每个线程执行一次，就将 count 计数加 1，当执行完以后，打印 count 的值。

package atomic;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

import annotation.NotThreadSafe;

@NotThreadSafe
public class NotThreadSafeConcurrency {

    private static int CLIENT_COUNT = 5000;
    private static int THREAD_COUNT = 200;
    private static int count = 0;
    private static int[] values = new int[11];

    private static ExecutorService executorService = Executors.newCachedThreadPool();
    private final static Semaphore semaphore = new Semaphore(THREAD_COUNT);
    private final static CountDownLatch countDownLatch = new CountDownLatch(CLIENT_COUNT);

    public static void main(String[] args) throws Exception {
        testAtomicInt();
    }

    private static void testAtomicInt() throws Exception {
        for (int i = 0; i < CLIENT_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // count每加 1，进行减 1 计数
                countDownLatch.countDown();
            });
        }
        // 等待线程池所有任务执行结束
        countDownLatch.await();
        executorService.shutdown();
        System.out.println("ConcurrencyDemo:" + count);
    }

    private static void add() {
        count++;
    }
}
//----------------------------------执行结果-------------------------------------------
由于是非线程安全的，所以运行结果总是 <= 5000
ConcurrencyDemo:5000
ConcurrencyDemo:4999
ConcurrencyDemo:4991
ConcurrencyDemo:4997

线程安全的高并发实现 AtomicInteger

package concurrency;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

import annotation.NotThreadSafe;
import annotation.ThreadSafe;

@ThreadSafe
public class ThreadSafeConcurrency {

    private static int CLIENT_COUNT = 5000;
    private static int THREAD_COUNT = 200;
    private static AtomicInteger count = new AtomicInteger(0);

    private static ExecutorService executorService = Executors.newCachedThreadPool();
    private final static Semaphore semaphore = new Semaphore(THREAD_COUNT);
    private final static CountDownLatch countDownLatch = new CountDownLatch(CLIENT_COUNT);

    public static void main(String[] args) throws Exception {
        testAtomicInteger();
    }

    private static void testAtomicInteger() throws Exception {
        for (int i = 0; i < CLIENT_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // count每加 1，进行减 1 计数
                countDownLatch.countDown();
            });
        }
        // 等待线程池所有任务执行结束
        countDownLatch.await();
        executorService.shutdown();
        System.out.println("ConcurrencyDemo:" + count);
    }

    private static void add() {
        count.incrementAndGet();
    }
}

//----------------------------------执行结果-------------------------------------------
由于是线程安全的，所以运行结果总是 == 5000
ConcurrencyDemo:5000
ConcurrencyDemo:5000
ConcurrencyDemo:5000

AtomicInteger 保证原子性

在 JDK1.5 中新增 java.util.concurrent(J.U.C) 包，它建立在 CAS 之上。CAS 是非阻塞算法的一种常见实现，相对于 synchronized 这种阻塞算法，它的性能更好。

CAS

CAS 就是 Compare and Swap 的意思，比较并操作。很多的 CPU 直接支持 CAS 指令。CAS 是一项乐观锁技术，当多个线程尝试使用 CAS 同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。

CAS 操作包含三个操作数 —— 内存位置（V）、预期原值（A）和新值（B）。当且仅当预期值 A 和内存值 V 相同时，将内存值 V 修改为 B，否则什么都不做。

JDK1.5 中引入了底层的支持，在 int、long 和对象的引用等类型上都公开了 CAS 的操作，并且 JVM 把它们编译为底层硬件提供的最有效的方法，在运行 CAS 的平台上，运行时把它们编译为相应的机器指令。在 java.util.concurrent.atomic 包下面的所有的原子变量类型中，比如 AtomicInteger，都使用了这些底层的JVM支持为数字类型的引用类型提供一种高效的 CAS 操作。

在 CAS 操作中，会出现 ABA 问题。就是如果 V 的值先由 A 变成 B，再由 B 变成 A，那么仍然认为是发生了变化，并需要重新执行算法中的步骤。

有简单的解决方案：不是更新某个引用的值，而是更新两个值，包括一个引用和一个版本号，即使这个值由 A 变为 B，然后 B 变为 A，版本号也是不同的。

AtomicStampedReference 和 AtomicMarkableReference 支持在两个变量上执行原子的条件更新。AtomicStampedReference 更新一个 “对象-引用” 二元组，通过在引用上加上 “版本号”，从而避免 ABA 问题，AtomicMarkableReference 将更新一个“对象引用-布尔值”的二元组。

AtomicInteger 实现

AtomicInteger 是一个支持原子操作的 Integer 类，就是保证对 AtomicInteger 类型变量的增加和减少操作是原子性的，不会出现多个线程下的数据不一致问题。如果不使用 AtomicInteger，要实现一个按顺序获取的 ID，就必须在每次获取时进行加锁操作，以避免出现并发时获取到同样的 ID 的现象。

package java.util.concurrent.atomic;

public class AtomicInteger extends Number implements java.io.Serializable {
    private static final long serialVersionUID = 6214790243416807050L;
    // setup to use Unsafe.compareAndSwapInt for updates
    private static final Unsafe unsafe = Unsafe.getUnsafe();

    private volatile int value;

    public AtomicInteger(int initialValue) {
        value = initialValue;
    }
    public AtomicInteger() {
    }

    public final int get() {
        return value;
    }

    //compareAndSet()方法调用的 compareAndSwapInt() 方法是一个 native 方法。compareAndSet 传入的为执行方法时获取到的 value 属性值，update 为加 1 后的值， compareAndSet 所做的为调用 Sun 的 UnSafe 的 compareAndSwapInt 方法来完成，此方法为 native 方法。
    //compareAndSwapInt 基于的是 CPU 的 CAS 指令来实现的。所以基于 CAS 的操作可认为是无阻塞的，一个线程的失败或挂起不会引起其它线程也失败或挂起。并且由于 CAS 操作是 CPU 原语，所以性能比较好。
    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }

    //先获取到当前的 value 属性值，然后将 value 加 1，赋值给一个局部的 next 变量，然而，这两步都是非线程安全的，但是内部有一个死循环，不断去做 compareAndSet 操作，直到成功为止，也就是修改的根本在 compareAndSet 方法里面。
    public final int getAndIncrement() {
        for (;;) {
            int current = get();
            int next = current + 1;
            if (compareAndSet(current, next))
                return current;
        }
    }

    public final int getAndDecrement() {
        for (;;) {
            int current = get();
            int next = current - 1;
            if (compareAndSet(current, next))
                return current;
        }
    }
}

AtomicInteger 中还有 IncrementAndGet() 和 DecrementAndGet() 方法，他们的实现原理和上面的两个方法完全相同，区别是返回值不同，getAndIncrement() 和 getAndDecrement() 两个方法返回的是改变之前的值，即current。IncrementAndGet() 和 DecrementAndGet() 两个方法返回的是改变之后的值，即 next。

Atomic 延伸其它类

原子更新基本类型

使用原子的方式更新基本类型，Atomic 包提供了以下 3 个类。

AtomicBoolean：原子更新布尔类型。

AtomicInteger：原子更新整型。

AtomicLong：原子更新长整型。

原子更新数组

通过原子的方式更新数组里的某个元素，Atomic 包提供了以下 3 个类。
AtomicIntegerArray：原子更新整型数组里的元素。
AtomicLongArray：原子更新长整型数组里的元素。
AtomicReferenceArray：原子更新引用类型数组里的元素。

int[] 测试

package concurrency;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import annotation.NotThreadSafe;
import annotation.ThreadSafe;

@NotThreadSafe
public class ThreadSafeConcurrency {

    private static int CLIENT_COUNT = 5000;
    private static int THREAD_COUNT = 200;
    private static int[] values = new int[11];

    private static ExecutorService executorService = Executors.newCachedThreadPool();
    private final static Semaphore semaphore = new Semaphore(THREAD_COUNT);
    private final static CountDownLatch countDownLatch = new CountDownLatch(CLIENT_COUNT);

    public static void main(String[] args) throws Exception {
        testAtomicIntArray();
    }

    private static void testAtomicIntArray() throws Exception {
        for (int i = 0; i < CLIENT_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    for (int j = 0; j < 10; j++) {// 所有元素+1
                        values[j]++;
                    }
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // count每加 1，进行减 1 计数
                countDownLatch.countDown();
            });
        }
        // 等待线程池所有任务执行结束
        countDownLatch.await();
        executorService.shutdown();
        for (int i = 0; i < 10; i++) {
            System.out.print(values[i] + " ");
        }

    }
}
//----------------------------------执行结果-------------------------------------------
4999 4998 4999 4997 4997 4998 4999 4998 4997 4997

AtomicIntegerArray 测试

package concurrency;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerArray;
import java.util.logging.Logger;

import annotation.NotThreadSafe;
import annotation.ThreadSafe;

@ThreadSafe
public class ThreadSafeConcurrency {

    private static int CLIENT_COUNT = 5000;
    private static int THREAD_COUNT = 200;
    private static AtomicInteger count = new AtomicInteger(0);
    private static int[] values = new int[10];
    private static AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(values);

    private static ExecutorService executorService = Executors.newCachedThreadPool();
    private final static Semaphore semaphore = new Semaphore(THREAD_COUNT);
    private final static CountDownLatch countDownLatch = new CountDownLatch(CLIENT_COUNT);

    public static void main(String[] args) throws Exception {
        testAtomicIntegerArray();
    }

    private static void testAtomicIntegerArray() throws Exception {
        for (int i = 0; i < CLIENT_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    for (int j = 0; j < 10; j++) {// 所有元素+1
                        atomicIntegerArray.incrementAndGet(j);
                    }
                    semaphore.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // count每加 1，进行减 1 计数
                countDownLatch.countDown();
            });
        }
        // 等待线程池所有任务执行结束
        countDownLatch.await();
        executorService.shutdown();
        for (int i = 0; i < 10; i++) {
            System.out.print(atomicIntegerArray.get(i) + " ");
        }

    }
}
//----------------------------------执行结果-------------------------------------------
5000 5000 5000 5000 5000 5000 5000 5000 5000 5000

原子更新引用

原子更新基本类型的 AtomicInteger，只能更新一个变量，如果要原子更新多个变量，就需要使用这个原子更新引用类型提供的类。Atomic 包提供了以下 3 个类。
AtomicReference：原子更新引用类型。
AtomicReferenceFieldUpdater：原子更新引用类型里的字段。
AtomicMarkableReference：原子更新带有标记位的引用类型。

package concurrency;
import java.util.concurrent.atomic.AtomicReference;

@ThreadSafe
public class ThreadSafeConcurrency {
    private static AtomicReference<Integer> atomicUserRef = new AtomicReference<Integer>(0);
    public static void main(String[] args) throws Exception {
        testAtomicReference();
    }
    private static void testAtomicReference() throws Exception {
        atomicUserRef.compareAndSet(0, 2);
        atomicUserRef.compareAndSet(0, 1);
        atomicUserRef.compareAndSet(1, 3);
        atomicUserRef.compareAndSet(2, 4);
        atomicUserRef.compareAndSet(3, 5);
        System.out.println("ConcurrencyDemo:" + atomicUserRef.get().toString());
    }
}
//----------------------------------执行结果-------------------------------------------
ConcurrencyDemo:4

原子更新字段

如果需原子地更新某个类里的某个字段时，就需要使用原子更新字段类，Atomic 包提供了以下 3 个类进行原子字段更新。
AtomicIntegerFieldUpdater：原子更新整型的字段的更新器。
AtomicLongFieldUpdater：原子更新长整型字段的更新器。
AtomicStampedReference：原子更新带有版本号的引用类型。该类将整数值与引用关联起来，可用于原子的更新数据和数据的版本号，可以解决使用 CAS 进行原子更新时可能出现的 ABA 问题。

要想原子地更新字段类需要两步。

第一步，因为原子更新字段类都是抽象类，每次使用的时候必须使用静态方法 newUpdater() 创建一个更新器，并且需要设置想要更新的类和属性。

第二步，更新类的字段（属性）必须使用 public volatile 修饰符。

package concurrency;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

@ThreadSafe
public class ThreadSafeConcurrency {
    private static AtomicIntegerFieldUpdater<User> atomicIntegerFieldUpdater = AtomicIntegerFieldUpdater.newUpdater(User.class, "old");
    private static User user;

    public static void main(String[] args) throws Exception {
        testAtomicIntegerFieldUpdater();
    }

    private static void testAtomicIntegerFieldUpdater() throws Exception {
        user = new User("user ", 100);
        atomicIntegerFieldUpdater.incrementAndGet(user);
        // 等待线程池所有任务执行结束
        System.out.println("ConcurrencyDemo:" + atomicIntegerFieldUpdater.get(user));
    }
}

class User {
    private String name;
    public volatile int old;//必须使用 volatile 标识，并且是 非 static
    public User(String name, int old) {
        this.name = name;
        this.old = old;
    }
    public String getName() {
        return name;
    }
    public int getOld() {
        return old;
    }
}
//----------------------------------执行结果-------------------------------------------
ConcurrencyDemo:101

最后编辑于：2018.07.10 18:19:52

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 212,185评论 6赞 493
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 90,445评论 3赞 385
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 157,684评论 0赞 348
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 56,564评论 1赞 284
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 65,681评论 6赞 386
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,874评论 1赞 290
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,025评论 3赞 408
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,761评论 0赞 268
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,217评论 1赞 303
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,545评论 2赞 327
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,694评论 1赞 341
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,351评论 4赞 332
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,988评论 3赞 315
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,778评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,007评论 1赞 266
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,427评论 2赞 360
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,580评论 2赞 349