JUC-线程安全集合类

一、概述

在 Java 中线程安全集合类可以分为三大类。

1. 遗留的线程安全集合：Hashtable、Vector；
2. Collections 装饰的线程安全集合：

3. java.util.concurrent.* 下的线程安全集合类，其中包含的又有以下三类关键词：

• Blocking：大部分实现基于锁，并提供用来阻塞的方法。

• CopyOnWrite：之类容器修改开销相对较重。

• Concurrent：类型的容器，内部很多操作使用 cas 优化，一般可以提供较高吞吐量，另外它还有弱一致性的特点：

  • 遍历时弱一致性，例如，当利用迭代器遍历时，如果容器发生修改，迭代器仍然可以继续进行遍历，这时内容是旧的；
  • 求大小弱一致性，size 操作未必是 100% 准确；
  • 读取弱一致性。

二、ConcurrentHashMap

2.1 并发问题

• 代码示例：

public class Demo {

    private static final String ALPHA = "abcedfghijklmnopqrstuvwxyz";
    private static final List<String> TEST_DATA = new ArrayList<>();

    static {
        int length = ALPHA.length();
        int count = 100;
        int capacity = length * count;
        List<String> list = new ArrayList<>(capacity);
        for (int i = 0; i < length; i++) {
            // 拿到每个字母。
            char c = ALPHA.charAt(i);
            for (int j = 0; j < count; j++) {
                // 添加字母各一百次到集合中。
                list.add(c + "");
            }
        }
        // 将集合中元素随机重排序。
        Collections.shuffle(list);
        TEST_DATA.addAll(list);
    }


    /**
     * 获取计数结果。
     *
     * @param threadNum 线程数。
     * @param sup       供给型函数式接口。
     * @param cons      消费型函数式接口。
     */
    private static <V> void getCount(
            int threadNum,
            Supplier<Map<String, V>> sup,
            BiConsumer<Map<String, V>, List<String>> cons
    ) {
        Map<String, V> map = sup.get();
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < threadNum; i++) {
            Thread t = new Thread(() -> cons.accept(map, TEST_DATA));
            threads.add(t);
        }

        threads.forEach(Thread::start);
        threads.forEach(t -> {
            try {
                t.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }
        });

        System.out.println(map);
    }

    /**
     * 方式一：使用非线程安全的 `map` 并发读取。
     */
    @Test
    public void test02() {
        getCount(
                2,

                () -> new HashMap<String, Integer>(),

                (map, words) -> {
                    for (String word : words) {
                        Integer counter = map.get(word);
                        int newValue = counter == null ? 1 : counter + 1;
                        map.put(word, newValue);
                    }
                }
        );
        // {a=193, b=192, c=196, d=197, e=193, f=193, g=195, h=195, i=193,
        // j=192, k=196, l=193, m=191, n=192, o=198, p=192, q=197, r=193,
        // s=192, t=175, u=194, v=193, w=193, x=195, y=191, z=195}
    }

    /**
     * 方式二：使用 `ConcurrentHashMap` 并发读取。
     */
    @Test
    public void test03() {
        getCount(
                2,

                () -> new ConcurrentHashMap<String, Integer>(),

                (map, words) -> {
                    for (String word : words) {
                        // 函数式编程，无需原子变量。
                        map.merge(word, 1, Integer::sum);
                    }
                }
        );
        // {a=200, b=200, c=200, d=200, e=200, f=200, g=200, h=200, i=200,
        // j=200, k=200, l=200, m=200, n=200, o=200, p=200, q=200, r=200,
        // s=200, t=200, u=200, v=200, w=200, x=200, y=200, z=200}
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

2.2 重要属性和内部类

备注：此处示例为 JDK 8 版本，数组简称（table），链表简称（bin）。

// 默认为 0
// 当初始化时, 为 -1
// 当扩容时, 为 -(1 + 扩容线程数)
// 当初始化或扩容完成后，为 下一次的扩容的阈值大小
private transient volatile int sizeCtl;

// 整个 ConcurrentHashMap 就是一个 Node[]
static class Node<K,V> implements Map.Entry<K,V> {}

// hash 表
transient volatile Node<K,V>[] table;

// 扩容时的 新 hash 表
private transient volatile Node<K,V>[] nextTable;

// 扩容时如果某个 bin 迁移完毕, 用 ForwardingNode 作为旧 table bin 的头结点
static final class ForwardingNode<K,V> extends Node<K,V> {}

// 用在 compute 以及 computeIfAbsent 时, 用来占位, 计算完成后替换为普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}

// 作为 treebin 的头节点, 存储 root 和 first
static final class TreeBin<K,V> extends Node<K,V> {}

// 作为 treebin 的节点, 存储 parent, left, right
static final class TreeNode<K,V> extends Node<K,V> {}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

2.3 重要方法

// 获取 Node[] 中第 i 个 Node
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i)
 
// cas 修改 Node[] 中第 i 个 Node 的值, c 为旧值, v 为新值
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v)
 
// 直接修改 Node[] 中第 i 个 Node 的值, v 为新值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v)
1
2
3
4
5
6
7
8

2.4 构造器

实现了懒惰初始化，在构造方法中仅仅计算了 table 大小，之后会在第一次使用时才会真正的创建。

    /**
     * Creates a new, empty map with an initial table size based on
     * the given number of elements ({@code initialCapacity}), table
     * density ({@code loadFactor}), and number of concurrently
     * updating threads ({@code concurrencyLevel}).
     *
     * @param initialCapacity the initial capacity. The implementation
     * performs internal sizing to accommodate this many elements,
     * given the specified load factor.
     * @param loadFactor the load factor (table density) for
     * establishing the initial table size
     * @param concurrencyLevel the estimated number of concurrently
     * updating threads. The implementation may use this value as
     * a sizing hint.
     * @throws IllegalArgumentException if the initial capacity is
     * negative or the load factor or concurrencyLevel are
     * nonpositive
     */
    public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins
            initialCapacity = concurrencyLevel;   // as estimated threads
        // tableSizeFor 保证计算大小是(2^n)：即16、32、64...
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        this.sizeCtl = cap;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

2.5 get 流程

    /**
     * Returns the value to which the specified key is mapped,
     * or {@code null} if this map contains no mapping for the key.
     *
     * 
More formally, if this map contains a mapping from a key
     * {@code k} to a value {@code v} such that {@code key.equals(k)},
     * then this method returns {@code v}; otherwise it returns
     * {@code null}.  (There can be at most one such mapping.)
     *
     * @throws NullPointerException if the specified key is null
     */
    public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        // spread 方法能确保返回结果是正数。
        int h = spread(key.hashCode());
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            // 如果头节点已经是要查找的 key。
            if ((eh = e.hash) == h) {
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            // hash 为负数表示该 bin 在扩容中或是 treebin, 这时调用 find 方法来查找。
            else if (eh < 0)
                return (p = e.find(h, key)) != null ? p.val : null;
            // 正常遍历链表, 用 equals 比较。
            while ((e = e.next) != null) {
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))
                    return e.val;
            }
        }
        return null;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

2.6 put 流程

• put()：

    /**
     * Maps the specified key to the specified value in this table.
     * Neither the key nor the value can be null.
     *
     * 
The value can be retrieved by calling the {@code get} method
     * with a key that is equal to the original key.
     *
     * @param key key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     * @return the previous value associated with {@code key}, or
     *         {@code null} if there was no mapping for {@code key}
     * @throws NullPointerException if the specified key or value is null
     */
    public V put(K key, V value) {
        return putVal(key, value, false); // *** putVal() 具体方法说明在下方 ***
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• putVal()：

    /** Implementation for put and putIfAbsent */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        // 其中 spread 方法会综合高位低位, 具有更好的 hash 性。
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            // f 是链表头节点；
            // fh 是链表头结点的 hash；
            // i 是链表在 table 中的下标。
            Node<K,V> f; int n, i, fh;
            // 要创建 table。
            if (tab == null || (n = tab.length) == 0)
                // 初始化 table 使用了 cas, 无需 synchronized 创建成功, 进入下一轮循环。 
                tab = initTable();  // *** initTable() 具体方法说明在下方 ***
            // 要创建链表头节点。
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                // 添加链表头使用了 cas, 无需 synchronized。
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            // 帮忙扩容。
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                // 锁住链表头节点。
                synchronized (f) {
                    // 再次确认链表头节点没有被移动。
                    if (tabAt(tab, i) == f) {
                        if (fh >= 0) {
                            binCount = 1;
                            // 遍历链表。
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                // 找到相同的key。
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    // 值更新。
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                // 已经是最后的节点了, 新增 Node, 追加至链表尾。
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        // 红黑树。
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            // putTreeVal 会看 key 是否已经在树中, 是, 则返回对应的 TreeNode。
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                    // 释放链表头节点的锁。
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        // 如果链表长度 >= 树化阈值(8), 进行链表转为红黑树。
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 增加 size 计数。
        addCount(1L, binCount);  // *** addCount() 具体方法说明在下方 ***
        return null;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

• initTable()：

    /**
     * Initializes table, using the size recorded in sizeCtl.
     */
    private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            // 尝试将 sizeCtl 设置为 -1（表示初始化 table）。
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                // 获得锁, 创建 table, 这时其它线程会在 while() 循环中 yield 直至 table 创建。
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

• addCount()：

    /**
     * Adds to count, and if table is too small and not already
     * resizing, initiates transfer. If already resizing, helps
     * perform transfer if work is available.  Rechecks occupancy
     * after a transfer to see if another resize is already needed
     * because resizings are lagging additions.
     *
     * @param x the count to add
     * @param check if <0, don't check resize, if <= 1 only check if uncontended
     */
	// check 是之前 binCount 的个数。
    private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        // 已经有了 counterCells, 向 cell 累加。
        if ((as = counterCells) != null ||
            // 还没有, 向 baseCount 累加。
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            // 还没有 counterCells。
            if (as == null || (m = as.length - 1) < 0 ||
                // 还没有 cell。
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                // cell cas 增加计数失败。
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                // 创建累加单元数组和cell, 累加重试。
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            // 获取元素个数。
            s = sumCount();
        }
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                int rs = resizeStamp(n);
                if (sc < 0) {
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    // newtable 已经创建了，帮忙扩容。
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                // 需要扩容，这时 newtable 未创建。
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

2.7 size 计算流程

• size 计算实际发生在 put，remove 改变集合元素的操作之中。
• 没有竞争发生，向 baseCount 累加计数。
• 有竞争发生，新建 counterCells，向其中的一个 cell 累加计数：
  • counterCells 初始有两个 cell。
  • 如果计数竞争比较激烈，会创建新的 cell 来累加计数。

    /**
     * {@inheritDoc}
     */
    public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }

    final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        // 将 baseCount 计数与所有 cell 计数累加。
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Java 8 数组（Node） +（ 链表 Node | 红黑树 TreeNode ） 以下数组简称（table），链表简称（bin）:

• 初始化：使用 cas 来保证并发安全，懒惰初始化 table；
• 树化：当 table.length < 64 时，先尝试扩容，超过 64 时，并且 bin.length > 8 时，会将链表树化，树化过程会用 synchronized 锁住链表头；
• put：如果该 bin 尚未创建，只需要使用 cas 创建 bin；如果已经有了，锁住链表头进行后续 put 操作，元素添加至 bin 的尾部；
• get：无锁操作仅需要保证可见性，扩容过程中 get 操作拿到的是 ForwardingNode 它会让 get 操作在新 table 进行搜索；
• 扩容：扩容时以 bin 为单位进行，需要对 bin 进行 synchronized，但这时妙的是其它竞争线程也不是无事可做，它们会帮助把其它 bin 进行扩容，扩容时平均只有 1/6 的节点会把复制到新 table 中；
• size：元素个数保存在 baseCount 中，并发时的个数变动保存在 CounterCell[] 当中。最后统计数量时累加即可。

2.8 对比 JDK 7 的 ConcurrentHashMap 有什么区别？

• JDK8 中新增了红黑树；
• JDK7中使用的是头插法，JDK8中使用的是尾插法；
• JDK7中使用了分段锁，而JDK8中没有使用分段锁了；
• JDK7中使用了 ReentrantLock，JDK8中没有使用 ReentrantLock 了，而使用了 Synchronized；
• JDK7中的扩容是每个 Segment 内部进行扩容，不会影响其他 Segment，而JDK8中的扩容和 HashMap 的扩容类似，只不过支持了多线程扩容，并且保证了线程安全。
• 补充说明：JDK6 / 7中的 ConcurrentHashMap 主要使用 Segment 来实现减小锁粒度，把 HashMap 分割成若干个 Segment ，在 put 的时候需要锁住Segment ，get 时候不加锁，使用 volatile 来保证可见性，当要统计全局时（比如 size），首先会尝试多次计算 modcount 来确定，这几次尝试中，是否有其他线程进行了修改操作，如果没有，则直接返回 size。如果有，则需要依次锁住所有的 Segment 来计算。JDK7中 ConcurrentHashmap 中，当长度过长碰撞会很频繁，链表的增改删查操作都会消耗很长的时间，影响性能。所以 JDK8 中完全重写了ConcurrentHashMap，代码量从原来的1000多行变成了 6000多行，实现上也和原来的分段式存储有很大的区别。

三、LinkedBlockingQueue

3.1 入队出队

• 入队源码：

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
  
      static class Node<E> {
        E item;

        /**
         * One of:
         * - the real successor Node
         * - this Node, meaning the successor is head.next
         * - null, meaning there is no successor (this is the last node)
         */
        Node<E> next;

        Node(E x) { item = x; }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

• 初始化链表：last = head = new Node(null); Dummy 节点用来占位，item 为 null。

• 当一个节点入队 last = last.next = node;

• 再来一个节点入队 last = last.next = node;

• 出队源码：

    /**
     * Removes a node from head of queue.
     *
     * @return the node
     */
    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• 示意图：

3.2 加锁分析

巧妙之处在于用了两把锁和 dummy （哑元）节点。

• 用一把锁，同一时刻，最多只允许有一个线程（生产者或消费者，二选一）执行。

• 用两把锁，同一时刻，可以允许两个线程同时（一个生产者与一个消费者）执行：

  • 消费者与消费者线程仍然串行。

  • 生产者与生产者线程仍然串行。

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();
1
2
3
4
5

3.3 线程安全分析

• 当节点总数大于 2 时（包括 dummy 节点），putLock 保证的是 last 节点的线程安全，takeLock 保证的是 head 节点的线程安全。两把锁保证了入队和出队没有竞争。

• 当节点总数等于 2 时（即一个 dummy 节点，一个正常节点）这时候，仍然是两把锁锁两个对象，不会竞争。

• 当节点总数等于 1 时（就一个 dummy 节点）这时 take 线程会被 notEmpty 条件阻塞，有竞争，会阻塞。

3.4 源码分析

• put()：

    /**
     * Inserts the specified element at the tail of this queue, waiting if
     * necessary for space to become available.
     *
     * @throws InterruptedException {@inheritDoc}
     * @throws NullPointerException {@inheritDoc}
     */
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
      	// holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
       	// count 用来维护元素计数。
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
           	// 满了就等待。
            while (count.get() == capacity) {
                notFull.await();
            }
           	// 有空位, 入队且计数加一。
            enqueue(node);
            c = count.getAndIncrement();
           	// 除了自己 put 以外, 队列还有空位, 由自己叫醒其他 put 线程。
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
       	// 如果队列中有一个元素, 叫醒 take 线程。
        if (c == 0)
           // 这里调用的是 notEmpty.signal() 而不是 notEmpty.signalAll() 是为了减少竞争。
            signalNotEmpty();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

• take() ：

    public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
      	// 如果队列中只有一个空位时, 叫醒 put 线程。
 				// 如果有多个线程进行出队, 第一个线程满足 c == capacity, 但后续线程 c < capacity。
        if (c == capacity)
          	// 这里调用的是 notFull.signal() 而不是 notFull.signalAll() 是为了减少竞争。
            signalNotFull();
        return x;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

3.5 性能比较

主要列举 LinkedBlockingQueue 与 ArrayBlockingQueue 的性能比较：

• Linked 支持有界，Array 强制有界。

• Linked 实现是链表，Array 实现是数组。

• Linked 是懒惰的，而 Array 需要提前初始化 Node 数组。

• Linked 每次入队会生成新 Node，而 Array 的 Node 是提前创建好的。

• Linked 两把锁，Array 一把锁。

四、ConcurrentLinkedQueue

4.1 概述

ConcurrentLinkedQueue 的设计与 LinkedBlockingQueue 非常像。

• 也是两把锁，同一时刻，可以允许两个线程同时（一个生产者与一个消费者）执行；

• dummy 节点的引入让两把锁将来锁住的是不同对象，避免竞争；

• 只是这锁使用了 cas 来实现。

4.2 简单实现

public class ConcurrentLinkedQueueTests {

    public static void main(String[] args) {
        MyQueue<String> queue = new MyQueue<>();
        queue.offer("3");
        queue.offer("2");
        queue.offer("1");
        System.out.println(queue);
        // 3->2->1->null
    }
}

class MyQueue<E> implements Queue<E> {

    private volatile Node<E> head;
    private volatile Node<E> last;

    public MyQueue() {
        head = last = new Node<>(null, null);
    }

    private static class Node<E> {
        
        volatile E item;

        public Node(E item, Node<E> next) {
            this.item = item;
            this.next = new AtomicReference<>(next);
        }

        AtomicReference<Node<E>> next;

    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (Node<E> p = head; p != null; p = p.next.get()) {
            E item = p.item;
            if (item != null) {
                sb.append(item).append("->");
            }
        }
        sb.append("null");
        return sb.toString();
    }

    @Override
    public boolean offer(E e) {

        Node<E> n = new Node<>(e, null);
        while (true) {
            // 获取尾节点
            AtomicReference<Node<E>> next = last.next;
            // S1: 真正尾节点的 next 是 null, cas 从 null 到新节点
            if (next.compareAndSet(null, n)) {
                // 这时的 last 已经是倒数第二, next 不为空了, 其它线程的 cas 肯定失败
                // S2: 更新 last 为倒数第一的节点
                last = n;
                return true;
            }
        }
    }

    @Override
    public int size() {
        return 0;
    }

    @Override
    public boolean isEmpty() {
        return false;
    }

    @Override
    public boolean contains(Object o) {
        return false;
    }

    @Override
    public Iterator<E> iterator() {
        return null;
    }

    @Override
    public Object[] toArray() {
        return new Object[0];
    }

    @Override
    public <T> T[] toArray(T[] a) {
        return null;
    }

    @Override
    public boolean add(E e) {
        return false;
    }

    @Override
    public boolean remove(Object o) {
        return false;
    }

    @Override
    public boolean containsAll(Collection<?> c) {
        return false;
    }

    @Override
    public boolean addAll(Collection<? extends E> c) {
        return false;
    }

    @Override
    public boolean removeAll(Collection<?> c) {
        return false;
    }

    @Override
    public boolean retainAll(Collection<?> c) {
        return false;
    }

    @Override
    public void clear() {

    }

    @Override
    public E remove() {
        return null;
    }

    @Override
    public E poll() {
        return null;
    }

    @Override
    public E element() {
        return null;
    }

    @Override
    public E peek() {
        return null;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

五、CopyOnWriteArrayList

CopyOnWriteArraySet 是它的马甲 底层实现采用了 写入时拷贝 的思想，增删改操作会将底层数组拷贝一份，更改操作在新数组上执行，这时不影响其它线程的并发读，读写分离。

5.1 源码分析

• 此处以新增为例：

    /**
     * Appends the specified element to the end of this list.
     *
     * @param e element to be appended to this list
     * @return {@code true} (as specified by {@link Collection#add})
     */
    public boolean add(E e) {
      	// 可重入锁。
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
          	// 获取旧的数组。
            Object[] elements = getArray();
            int len = elements.length;
          	// 拷贝新的数组（这里是比较耗时的操作，但不影响其它读线程）。
            Object[] newElements = Arrays.copyOf(elements, len + 1);
          	// 添加新元素。
            newElements[len] = e;
          	// 替换旧的数组。
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

• 其他读操作并未加锁（适合读多写少场景），此处以遍历读取为例：

    public void forEach(Consumer<? super E> action) {
        if (action == null) throw new NullPointerException();
        Object[] elements = getArray();
        int len = elements.length;
        for (int i = 0; i < len; ++i) {
            @SuppressWarnings("unchecked") E e = (E) elements[i];
            action.accept(e);
        }
    }
1
2
3
4
5
6
7
8
9

5.2 弱一致性

• get() 及 迭代器都有弱一致性的特点。
• 此处以迭代器为例：

public class CopyOnWriteArrayListTests {
    public static void main(String[] args) throws InterruptedException {
        CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Iterator<Integer> iter = list.iterator();

        new Thread(() -> {
            list.remove(0);
            list.forEach(System.out::print);
            // 2 3
            System.out.println();
        }).start();

        TimeUnit.SECONDS.sleep(1);

        while (iter.hasNext()) {
            System.out.print(iter.next());
            // 1 2 3
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 注意（并不是弱一致性就不好）：
  • 数据库的 MVCC 都是弱一致性的表现。
  • 并发高和一致性是矛盾的，需要权衡。

六、结束语

---

“-------怕什么真理无穷,进一寸有一寸的欢喜。”

微信公众号搜索：饺子泡牛奶。