十六、JUC下常用类(包含源码) - 第二部分

1. StampedLock

该类自 JDK 8 加入，是为了进一步优化读性能，它的特点是在使用读锁、写锁时都必须配合【戳】使用
加解读锁

long stamp = lock.readLock();lock.unlockRead(stamp);

加解写锁

long stamp = lock.writeLock();lock.unlockWrite(stamp);

乐观读，StampedLock 支持 tryOptimisticRead() 方法（乐观读），读取完毕后需要做一次 戳校验 如果校验通过，表示这期间确实没有写操作，数据可以安全使用，如果校验没通过，需要重新获取读锁，保证数据安全。

long stamp = lock.tryOptimisticRead();// 验戳if(!lock.validate(stamp)){// 锁升级(乐观读锁升级为读锁)}

例子：

public int read(int readTime) {long stamp = lock.tryOptimisticRead();log.debug("optimistic read locking...{}", stamp);sleep(readTime);if (lock.validate(stamp)) {log.debug("read finish...{}, data:{}", stamp, data);return data;}// 锁升级 - 读锁log.debug("updating to read lock... {}", stamp);try {stamp = lock.readLock();log.debug("read lock {}", stamp);sleep(readTime);log.debug("read finish...{}, data:{}", stamp, data);return data;} finally {log.debug("read unlock {}", stamp);lock.unlockRead(stamp);}}public void write(int newData) {long stamp = lock.writeLock();log.debug("write lock {}", stamp);try {sleep(2);this.data = newData;} finally {log.debug("write unlock {}", stamp);lock.unlockWrite(stamp);}}

注意：

• StampedLock 不支持条件变量
• StampedLock 不支持可重入

2. Semaphore

Semaphore 信号量，用来限制能同时访问共享资源的线程上限。

举例

public class TestSemaphore {    public static void main(String[] args) { // 1. 创建 semaphore 对象，并规定大小为 3 Semaphore semaphore = new Semaphore(3); // 创建10个线程同时运行 for (int i = 0; i < 10; i++) {     new Thread(()->{  // 加锁  try {      semaphore.acquire();  } catch (InterruptedException e) {      e.printStackTrace();  }  try {      System.out.println("running..." + new Date());      try {   Thread.sleep(1000);      } catch (InterruptedException e) {   e.printStackTrace();      }      System.out.println("end..." + new Date());  } finally {      semaphore.release();  }     }).start(); }    }}

加锁解锁流程

Semaphore 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后停车场显示空余车位减一

刚开始，permits（state）为 3，这时 5 个线程来获取资源

假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，进入 AQS 队列park 阻塞

这时 Thread-4 释放了 permits，状态如下

接下来 Thread-0 竞争成功，permits 再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态

1. 构造方法，设置信号量原理

public Semaphore(int permits) { sync = new NonfairSync(permits);    }// 调用到，传给父类NonfairSync(int permits) { super(permits);    }

abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 1192457210091910933L; Sync(int permits) {     setState(permits); // 发现就是将传入的信号量赋值给了state变量 }......}

2. acquire() 原理

public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1);    }

public final void acquireSharedInterruptibly(int arg)     throws InterruptedException { if (Thread.interrupted())     throw new InterruptedException(); if (tryAcquireShared(arg) < 0)     doAcquireSharedInterruptibly(arg);    }

进入Semaphore内部非公平的实现

protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires);    }// 这里会将信号量(state一直减一，直到减为0，就不能在有更多的线程获取锁了)final int nonfairTryAcquireShared(int acquires) { for (;;) {     int available = getState();     int remaining = available - acquires;  // state一直减一     // 当state为0，remaining为-1；会跳过CAS操作，不允许更多线程加入     if (remaining < 0 ||     compareAndSetState(available, remaining))  return remaining; }    }

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); boolean failed = true; try {     for (;;) {  final Node p = node.predecessor();  if (p == head) {  // tryAcquireShared()内部也就是调用nonfairTryAcquireShared()      int r = tryAcquireShared(arg);if (r >= 0) {   setHeadAndPropagate(node, r);   p.next = null; // help GC   failed = false;   return;      }  } // 获取不到锁，到AQS队列中park  if (shouldParkAfterFailedAcquire(p, node) &&      parkAndCheckInterrupt())      throw new InterruptedException();     } } finally {     if (failed)  cancelAcquire(node); }    }

被唤醒之后，并且获取到了锁，进入setHeadAndPropagate方法，修改头结点，unpark紧跟在后面的一个共享Node。

private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below setHead(node); if (propagate > 0 || h == null || h.waitStatus < 0 ||     (h = head) == null || h.waitStatus < 0) {     Node s = node.next;     if (s == null || s.isShared())  doReleaseShared(); }    }

private void doReleaseShared() { for (;;) {     Node h = head;     if (h != null && h != tail) {  int ws = h.waitStatus;  if (ws == Node.SIGNAL) {      if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))   continue;     // loop to recheck cases      unparkSuccessor(h);  }  else if (ws == 0 &&    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))      continue;  // loop on failed CAS     }     if (h == head)     // loop if head changed  break; }    }

3. release()原理

public void release() { sync.releaseShared(1);    }

public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {     doReleaseShared();     return true; } return false;    }

进入Semaphore内部非公平的实现

protected final boolean tryReleaseShared(int releases) { for (;;) {     int current = getState();     int next = current + releases;  // state加1     if (next < current) // overflow  throw new Error("Maximum permit count exceeded");     if (compareAndSetState(current, next))  // CAS修改state  return true; }    }

private void doReleaseShared() { for (;;) {     Node h = head;     if (h != null && h != tail) {  int ws = h.waitStatus;  if (ws == Node.SIGNAL) {  // 将Node状态从-1 CAS 修改为0      if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))   continue;     // loop to recheck cases      // 成功 unpark 头结点后的节点，该节点线程再次尝试获取锁      unparkSuccessor(h);    }  else if (ws == 0 &&    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))      continue;  // loop on failed CAS     }     if (h == head)     // loop if head changed  break; }    }

private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0)     compareAndSetWaitStatus(node, ws, 0); Node s = node.next; if (s == null || s.waitStatus > 0) {     s = null;     for (Node t = tail; t != null && t != node; t = t.prev)  if (t.waitStatus <= 0)      s = t; } if (s != null)     LockSupport.unpark(s.thread);    }

3. CountdownLatch

用来进行线程同步协作，等待所有线程完成倒计时。
其中构造参数用来初始化等待计数值，await() 用来等待计数归零，countDown() 用来让计数减一

应用：游戏载入

public class TestCountDownLatch {    public static void main(String[] args) throws InterruptedException { ExecutorService service = Executors.newFixedThreadPool(10); CountDownLatch latch = new CountDownLatch(10); Random r = new Random(); String[] player = new String[10]; for (int i = 0; i < 10; i++) {     int k = i;     service.submit(()->{  for (int j = 0; j <= 100; j++) {      try {   // 模拟延迟效果   Thread.sleep(r.nextInt(100));      } catch (InterruptedException e) {   e.printStackTrace();      }      player[k] = j + "%";  // 玩家载入      System.out.print("\r" + Arrays.toString(player));  }  // 当前玩家载入完成  latch.countDown();     }); } latch.await(); System.out.println("\n玩家全部载入，游戏开始"); service.shutdown();    }}

构造方法

// 构造方法，将传入值设置给statepublic CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count);    }

常用方法

// 等待方法public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1);    }// 计数减一方法public void countDown() { sync.releaseShared(1);    }

public final void acquireSharedInterruptibly(int arg)     throws InterruptedException { if (Thread.interrupted())     throw new InterruptedException(); if (tryAcquireShared(arg) < 0)     doAcquireSharedInterruptibly(arg);  // 这里实现与前面的锁一致    }public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {     doReleaseShared();  // 这里实现与前面的锁一致     return true; } return false;    }

protected int tryAcquireShared(int acquires) {// state为0，不能继续获取，返回1// state不为0，还可以继续获取锁，返回-1 return (getState() == 0) ? 1 : -1;      }    protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) {     int c = getState();     if (c == 0)  return false;     int nextc = c-1;  // 计数减1     if (compareAndSetState(c, nextc))  return nextc == 0; }    }

4. CyclicBarrier

循环栅栏，用来进行线程协作，等待线程满足某个计数。构造时设置『计数个数』，每个线程执行到某个需要“同步”的时刻调用 await() 方法进行等待(计数减1)，当等待的线程数满足『计数个数』时(计数减为0)，继续执行，这时会将计数重置为计数个数，从而达到重复循环的使用

使用，模拟循环两个任务的执行

public class TestCyclicBarrier {    public static void main(String[] args) { ExecutorService service = Executors.newFixedThreadPool(2); CyclicBarrier barrier = new CyclicBarrier(2, ()->{     System.out.println("test1, test2 finish..."); }); for (int i = 0; i < 3; i++) {     service.submit(()->{  System.out.println("test1 START...");  try {      Thread.sleep(1000);      barrier.await();  // 2 - 1 = 1  } catch (InterruptedException | BrokenBarrierException e) {      e.printStackTrace();  }     });     service.submit(()->{  System.out.println("test2 START...");  try {      Thread.sleep(2000);      barrier.await();  // 1 - 1 = 0  } catch (InterruptedException | BrokenBarrierException e) {      e.printStackTrace();  }     }); } service.shutdown();    }}

注意最好将线程池数量和CyclicBarrier任务数保持一致，这样才能保证同步的正确性

十七、JUC中线程安全集合类

1. 介绍

线程安全集合类可以分为三大类：

1. 遗留的线程安全集合如 Hashtable ， Vector
2. 使用 Collections 装饰的线程安全集合，如：
   Collections.synchronizedCollection
   Collections.synchronizedList
   Collections.synchronizedMap
   Collections.synchronizedSet
   Collections.synchronizedNavigableMap
   Collections.synchronizedNavigableSet
   Collections.synchronizedSortedMap
   Collections.synchronizedSortedSet
3. java.util.concurrent.*

重点介绍java.util.concurrent. 下的线程安全集合类，可以发现它们有规律，里面包含三类关键词：
Blocking、CopyOnWrite、Concurrent*

1. Blocking 大部分实现基于锁，并提供用来阻塞的方法
2. CopyOnWrite 之类容器修改开销相对较重
3. Concurrent 类型的容器

内部很多操作使用 cas 优化，一般可以提供较高吞吐量
弱一致性

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

遍历时如果发生了修改，对于非安全容器来讲，使用 fail-fast 机制也就是让遍历立刻失败，抛出
ConcurrentModificationException，不再继续遍历

2. ConcurrentHashMap

1. JDK1.7前，HashMap存在死链问题

JDK1.7以前，添加，扩容时移动元素等操作采用头插法，会导致多线程情况下产生死链(循环链表问题)
死链问题图解
JDK1.8以后，上述操作改为使用尾插法，解决了这个问题，但是也依然存在数据丢失的问题

2. JDK1.8 ConcurrentHashMap 源码流程

1. 重要属性、内部类及方法

重要属性和内部类

// 默认为 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 时, 用来占位, 计算完成后替换为普通 Nodestatic final class ReservationNode<K,V> extends Node<K,V> {}// 作为 treebin 的头节点, 存储 root 和 firststatic final class TreeBin<K,V> extends Node<K,V> {}// 作为 treebin 的节点, 存储 parent, left, rightstatic final class TreeNode<K,V> extends Node<K,V> {}

重要方法

// 获取 Node[] 中第 i 个 Nodestatic 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)

2. 构造器分析

// initialCapacity 初始容量// loadFactor 负载因子// concurrencyLevel 并法量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 // 实现了懒惰初始化，在构造方法中仅仅计算了 table 的大小，以后在第一次使用时才会真正创建 long size = (long)(1.0 + (long)initialCapacity / loadFactor); // tableSizeFor方法 是保证计算的大小是 2 ^ n (16,32,64) int cap = (size >= (long)MAXIMUM_CAPACITY) ?     MAXIMUM_CAPACITY : tableSizeFor((int)size); // 设置真正的初始大小，初始化table[] 设置大小使用 this.sizeCtl = cap;    }

get流程

public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // soread 保证获取到的hash码为正数，因为负数表示正在扩容或者是树，之后要做判断 int h = spread(key.hashCode()); // 如果table不为空(已经创建)，并且 table长度大于0(有元素)  // 并且key的hash对应在的table中的链表不为空， 则进入if if ((tab = table) != null && (n = tab.length) > 0 &&     (e = tabAt(tab, (n - 1) & h)) != null) {     // 判断头结点是否为key     // hash 如果相等 进入     if ((eh = e.hash) == h) {     // 如果key相等，则直接返回头结点的value  if ((ek = e.key) == key || (ek != null && key.equals(ek)))      return e.val;     }     // 如果头结点的hash为负数     // 说明正在扩容(ForwardingNode) 或者 代表为一颗红黑树(TreeBin)     else if (eh < 0)     // 调用其find()，到新数组或树中查找值  return (p = e.find(h, key)) != null ? p.val : null;   // 上面都不满足，则进入这里正常循环遍历链表，查询链表里是否有对应的key     while ((e = e.next) != null) {  if (e.hash == h &&      ((ek = e.key) == key || (ek != null && key.equals(ek))))      return e.val;     } } // 查找失败，返回null return null;    }

头结点hash为-1，代表正在扩容

头结点hash为-2，代表为一颗红黑树

put流程

public V put(K key, V value) { return putVal(key, value, false);    }

// onlyIfAbsent： false代表相同key会替换value； true表示不会替换valuefinal V putVal(K key, V value, boolean onlyIfAbsent) {// ConcurrentHashMap 的键和值 不允许为空值 if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); // 获取key对应的正数hash // binCount如果大于0，代表有hash冲突，桶位置可能有新值加入，等于0则没有 int binCount = 0; for (Node<K,V>[] tab = table;;) { // f为链表头结点，fn时链表头节点的hash，i是链表在table中的下标     Node<K,V> f; int n, i, fh;     // 如果table没有被创建     if (tab == null || (n = tab.length) == 0)     // 初始化table使用了cas，无序synchronized，创建成功，进入下一轮循环  tab = initTable();     // 创建链表的头结点，如果table对应key的hash位置还为空     // 就创建新节点，将key，value放入，作为table该位置的头节点     else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {     // 添加链表头结点也是用了cas  if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))      break;     // no lock when adding to empty bin     }     // 帮忙扩容，如果头结点是ForwardingNode，说明别的线程正在扩容     else if ((fh = f.hash) == MOVED)     // 扩容是以链表为单位重新转移节点，所以会锁住其中一个链表，去帮助其他线程扩容  tab = helpTransfer(tab, f);     else {     // 当前table不用初始化，头结点有值且发生了哈西冲突，没有在扩容  V oldVal = null;  // 锁住这个桶的链表的头结点  synchronized (f) {  // 头结点没有被其他线程改变      if (tabAt(tab, i) == f) {      // 头结点hash值大于等于0，表示不是红黑树，是链表   if (fh >= 0) {binCount = 1; // 修改binCount的值大于0，表示存在hash冲突// 遍历桶位置的链表for (Node<K,V> e = f;; ++binCount) {    K ek;    // 找到了key，则更新value    if (e.hash == hash && ((ek = e.key) == key ||  (ek != null && key.equals(ek)))) { oldVal = e.val; // 更新value if (!onlyIfAbsent)     e.val = value; break;    }    // 没有找到key，则将新的key，value存入节点，尾插到链表后    Node<K,V> pred = e;    if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key,      value, null); break;    }}   }   // 头结点值小于0，代表是红黑树(不会是扩容，因为上面已经走过扩容判断了)   else if (f instanceof TreeBin) {Node<K,V> p;binCount = 2;  // // 修改binCount的值大于0，表示存在hash冲突// 将头结点转换为TreeBin，调用红黑树的查找方法if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,   value)) != null) {    oldVal = p.val;    // 更新    if (!onlyIfAbsent) p.val = value;}   }      }  }  // binCount如果大于0，表示发生了哈希冲突，可能有新节点加入  // 所以就要看看是否需要转为红黑树  if (binCount != 0) {      if (binCount >= TREEIFY_THRESHOLD)      // table容量到了64 且 链表长度 >= 树化阈值(8), 进行链表转为红黑树   treeifyBin(tab, i);      if (oldVal != null)   return oldVal;      break;  }     } } // 增加 size 计数，可能进行扩容操作 addCount(1L, binCount); return null;    }

put中initTable()流程

private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) {     if ((sc = sizeCtl) < 0)     // 其他线程执行完，再次循环查看是否创建成功，创建成功就不会在进入while循环了  Thread.yield(); // lost initialization race; just spin     // 尝试将 sizeCtl 设置为 -1(表示正在初始化table)     else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {     // 修改成功，获得锁，创建table     // 其他线程发现sizeCtl为-1，就会在while循环中yield，直至创建table成功  try {  // 第一次进入，判断成功      if ((tab = table) == null || tab.length == 0) {      // 设置table初始大小，默认为16   int n = (sc > 0) ? sc : DEFAULT_CAPACITY;   @SuppressWarnings("unchecked")   Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];   table = tab = nt;  // 初始化table   sc = n - (n >>> 2);  // 这里就是 容量 * 0.75      }  } finally {      sizeCtl = sc; // 将sizeCtl设置为扩容阈值  }  break;     } } return tab;    }

put中addCount()流程

// 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;     // 还没有累加单元数组，或者还没有累加单元，或者累加单元cas增加计数失败     if (as == null || (m = as.length - 1) < 0 ||  (a = as[ThreadLocalRandom.getProbe() & m]) == null ||  !(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;     // 如果s大于了 sizeCtl(扩容阈值)，且table不为空，且table长度不大于64，则尝试扩容     while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&     (n = tab.length) < MAXIMUM_CAPACITY) {  int rs = resizeStamp(n);  // 首次进入，sc为扩容阈值大于0；  // 多线程情况下，其他没有扩容的线程执行到这，发现sc被cas为负数，会帮忙扩容  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);  }  // 这里的cas会将sc修改为一个负数，防止重复扩容，使其他线程帮忙扩容  else if (U.compareAndSwapInt(this, SIZECTL, sc,   (rs << RESIZE_STAMP_SHIFT) + 2))      // 需要扩容，这时 newtable 未创建      transfer(tab, null);  s = sumCount();     } }    }

addCount中transfer()流程

size计算流程(不能保证正确，没有线程安全保护)

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; long sum = baseCount; // 获取baseCount  if (as != null) { // 累加单元数组不为空，遍历所有累加单元     for (int i = 0; i < as.length; ++i) {     // 如果累加单元中有值，则累加到sum中，返回  if ((a = as[i]) != null)      sum += a.value;     } } return sum;    }

3. JDK1.7 ConcurrentHashMap 源码流程

它维护了一个 segment 数组，每个 segment 对应一把锁

• 优点：如果多个线程访问不同的 segment，实际是没有冲突的，这与 jdk8 中是类似的
• 缺点：Segments 数组默认大小为16，这个容量初始化指定后就不能改变了，并且不是懒惰初始化

1. 构造器分析

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();if (concurrencyLevel > MAX_SEGMENTS)concurrencyLevel = MAX_SEGMENTS;// ssize 必须是 2^n, 即 2, 4, 8, 16 ... 表示了 segments 数组的大小int sshift = 0;int ssize = 1;while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// segmentShift 默认是 32 - 4 = 28this.segmentShift = 32 - sshift;// segmentMask 默认是 15 即 0000 0000 0000 1111this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;int cap = MIN_SEGMENT_TABLE_CAPACITY;while (cap < c)cap <<= 1;// 创建 segments and segments[0]Segment<K,V> s0 = new Segment<K,V>(loadFactor, (int)(cap * loadFactor),(HashEntry<K,V>[])new HashEntry[cap]);Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss;}

可以看到 ConcurrentHashMap 没有实现懒惰初始化，空间占用不友好
其中 this.segmentShift 和 this.segmentMask 的作用是决定将 key 的 hash 结果匹配到哪个 segment
例如，根据某一 hash 值求 segment 位置，先将高位向低位移动 this.segmentShift 位

结果再与 this.segmentMask 做位于运算，最终得到 1010 即下标为 10 的 segment

2. put流程

public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();int hash = hash(key);// 计算出 segment 下标int j = (hash >>> segmentShift) & segmentMask;// 获得 segment 对象, 判断是否为 null, 是则创建该 segmentif ((s = (Segment<K,V>)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE)) == null) {// 这时不能确定是否真的为 null, 因为其它线程也发现该 segment 为 null,// 因此在 ensureSegment 里用 cas 方式保证该 segment 安全性s = ensureSegment(j);}// 进入 segment 的put 流程return s.put(key, hash, value, false);}

segment 继承了可重入锁（ReentrantLock），它的 put 方法为

final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试加锁HashEntry<K,V> node = tryLock() ? null :// 如果不成功, 进入 scanAndLockForPut 流程// 如果是多核 cpu 最多 tryLock 64 次, 进入 lock 流程// 在尝试期间, 还可以顺便看该节点在链表中有没有, 如果没有顺便创建出来scanAndLockForPut(key, hash, value);// 执行到这里 segment 已经被成功加锁, 可以安全执行V oldValue;try {HashEntry<K,V>[] tab = table;int index = (tab.length - 1) & hash;HashEntry<K,V> first = entryAt(tab, index);for (HashEntry<K,V> e = first;;) {if (e != null) {// 更新K k;if ((k = e.key) == key ||(e.hash == hash && key.equals(k))) {oldValue = e.value;if (!onlyIfAbsent) {e.value = value;++modCount;}break;}e = e.next;}else {// 新增// 1) 之前等待锁时, node 已经被创建, next 指向链表头if (node != null)node.setNext(first);else// 2) 创建新 node， 头插法node = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;// 3) 扩容if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node);else// 将 node 作为链表头setEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}return oldValue;}

3. rehash 流程

发生在 put 中，因为此时已经获得了锁，因此 rehash 时不需要考虑线程安全

private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;int oldCapacity = oldTable.length;int newCapacity = oldCapacity << 1;  // 扩容一倍threshold = (int)(newCapacity * loadFactor);HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity];int sizeMask = newCapacity - 1;for (int i = 0; i < oldCapacity ; i++) {HashEntry<K,V> e = oldTable[i];if (e != null) {HashEntry<K,V> next = e.next;int idx = e.hash & sizeMask;if (next == null) // Single node on list 只有一个节点，直接搬迁到对应位置newTable[idx] = e;else { // Reuse consecutive sequence at same slotHashEntry<K,V> lastRun = e;int lastIdx = idx;// 过一遍链表, 尽可能把 rehash 后 idx 不变的节点重用for (HashEntry<K,V> last = next;last != null;last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}newTable[lastIdx] = lastRun;// 剩余节点需要新建for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {V v = p.value;int h = p.hash;int k = h & sizeMask;HashEntry<K,V> n = newTable[k];// 新建节点，放入扩容的链表中，(旧节点不变)newTable[k] = new HashEntry<K,V>(h, p.key, v, n);}}}}// 扩容完成, 才加入新的节点int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;// 替换为新的 HashEntry tabletable = newTable;}

4. get流程

get 时并未加锁，用了 UNSAFE 方法保证了可见性，扩容过程中，get 先发生就从旧表取内容，get 后发生就从新表取内容

public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);// u 为 segment 对象在数组中的偏移量long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;// s 即为 segmentif ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);e != null; e = e.next) {K k;if ((k = e.key) == key || (e.hash == h && key.equals(k)))return e.value;}}return null;}

5. size流程

• 计算元素个数前，先不加锁计算两次，如果前后两次结果如一样，认为个数正确返回
• 如果不一样，进行重试，重试次数超过 3，将所有 segment 锁住，重新计算个数返回

public int size() {// Try a few times to get accurate count. On failure due to// continuous async changes in table, resort to locking.final Segment<K,V>[] segments = this.segments;int size;boolean overflow; // true if size overflows 32 bitslong sum; // sum of modCountslong last = 0L; // previous sumint retries = -1; // first iteration isn't retrytry {for (;;) {if (retries++ == RETRIES_BEFORE_LOCK) {// 超过重试次数, 需要创建所有 segment 并加锁for (int j = 0; j < segments.length; ++j)ensureSegment(j).lock(); // force creation}sum = 0L;size = 0;overflow = false;for (int j = 0; j < segments.length; ++j) {Segment<K,V> seg = segmentAt(segments, j);if (seg != null) {sum += seg.modCount;int c = seg.count;if (c < 0 || (size += c) < 0)overflow = true;}}if (sum == last)break;last = sum;}} finally {if (retries > RETRIES_BEFORE_LOCK) {for (int j = 0; j < segments.length; ++j)segmentAt(segments, j).unlock();}}return overflow ? Integer.MAX_VALUE : size;}

3. LinkedBlockingQueue

1. 基本入队出队操作

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {static class Node<E> {E item;/* 下列三种情况之一* - 真正的后继节点* - 自己, 发生在出队时* - null, 表示是没有后继节点, 是最后了*/Node<E> next;Node(E x) { item = x; }}}

2. 加锁分析

高明之处在于用了两把锁和 dummy 节点，保证生产者和消费者同时可以执行

• 用一把锁，同一时刻，最多只允许有一个线程（生产者或消费者，二选一）执行
• 用两把锁，同一时刻，可以允许两个线程同时（一个生产者与一个消费者）执行
  消费者与消费者线程仍然串行
  生产者与生产者线程仍然串行

线程安全分析

• 当节点总数大于 2 时（包括 dummy 节点），putLock 保证的是 last 节点的线程安全，takeLock 保证的是 head 节点的线程安全。两把锁保证了入队和出队没有竞争
• 当节点总数等于 2 时（即一个 dummy 节点，一个正常节点）这时候，仍然是两把锁锁两个对象，不会竞争
• 当节点总数等于 1 时（就一个 dummy 节点）这时 take 线程会被 notEmpty 条件阻塞，有竞争，会阻塞

// 用于 put(阻塞) offer(非阻塞)private final ReentrantLock putLock = new ReentrantLock();// 用户 take(阻塞) poll(非阻塞)private final ReentrantLock takeLock = new ReentrantLock();

3. put操作

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 {// 队列满了，进入等待     while (count.get() == capacity) {     // 倒过来读就好: 等待 notFull  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();    }

4. take操作

public E take() throws InterruptedException { E x; int c = -1; // count 用来维护元素计数 final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { // 队列为空，进入等待     while (count.get() == 0) {  notEmpty.await();     }     // 队列不为空，出队且计数减一     x = dequeue();     // 等到原值并减一     c = count.getAndDecrement();     // 除了自己take以外，队列中还有值，由自己唤醒其他take线程，进行出队操作     if (c > 1)  notEmpty.signal(); } finally {     takeLock.unlock(); }// 如果队列中只有一个空位时, 叫醒 put 线程// 如果有多个线程进行出队, 第一个线程满足 c == capacity, 但后续线程 c < capacity if (c == capacity) // 这里调用的是 notFull.signal() 而不是 notFull.signalAll() 是为了减少竞争     signalNotFull(); return x;    }

由 put 唤醒 put 是为了避免信号不足

5. 性能比较

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

• Linked 支持有界，Array 强制有界
• Linked 实现是链表，Array 实现是数组
• Linked 是懒惰的，而 Array 需要提前初始化 Node 数组
• Linked 每次入队会生成新 Node，而 Array 的 Node 是提前创建好的
• Linked 两把锁，Array 一把锁

4. ConcurrentLinkedQueue

ConcurrentLinkedQueue 的设计与 LinkedBlockingQueue 非常像，也是

• 两把【锁】，同一时刻，可以允许两个线程同时（一个生产者与一个消费者）执行
• dummy 节点的引入让两把【锁】将来锁住的是不同对象，避免竞争
• 只是这【锁】使用了 cas 来实现

事实上，ConcurrentLinkedQueue 应用还是非常广泛的

例如之前讲的 Tomcat 的 Connector 结构时，Acceptor 作为生产者向 Poller 消费者传递事件信息时，正是采用了 ConcurrentLinkedQueue 将 SocketChannel 给 Poller 使用

5. CopyOnWriteArrayList

1. 写时复制

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

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(); }    }

其它读操作并未加锁，例如：

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); }    }

适合『读多写少』的应用场景

get 弱一致性

读取线程拿到的是旧数组，遍历；但此时新数组中，已经进行了remove操作，体现了弱一致性

迭代器弱一致性

CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();list.add(1);list.add(2);list.add(3);Iterator<Integer> iter = list.iterator(); // 迭代器获取的时原数组引用new Thread(() -> {// 线程中将 1 删除， 打印 [2,3]list.remove(0);System.out.println(list);}).start();sleep1s();// 因为迭代器中list是原数组引用，所以打印的还是原数组[1,2,3]while (iter.hasNext()) {System.out.println(iter.next());}

不要觉得弱一致性就不好
数据库的 MVCC 都是弱一致性的表现
并发高和一致性是矛盾的，需要权衡