[JCIP笔记]JDK并发包

Posted 2020-11-01 mskitten

这一节来讲一讲java.util.concurrent这个包里的一些重要的线程安全有关类。

synchronized容器

synchronized容器就是把自己的内部状态封装起来，通过把每一个public方法设置成同步来控制对共享变量的访问的容器。主要包括Vector, Hashtable，以及Collections.synchronizedxxx()方法提供的wrapper。

synchronized容器的问题-client locking

首先，synchronzied容器虽然是线程安全的，但是要访问容器内部数据的线程只能先拿到容器的内置锁才能访问，实际上相当于串行访问，CPU利用率和效率都不高。
另外还有一个值得注意的地方，就是用户代码使用synchronized容器时，如果需要做一些复合操作，比如put-if-absent，仍然要显式加锁（称为client locking），否则会产生race condition。
比如以下操作：

1 public Object getLast(Vector list){
2 　　int last = list.size() - 1; //1
3 　　return list.get(last); //2
4 }
5 public void removeLast(Vector list){
6 　　int last = list.size() - 1; //3
7 　　list.remove(last); //4
8 }

以上两个方法都对Vector进行了复合操作，在不加锁的情况下可能产生这样一种场景：线程A调用getLast()，同时线程B调用removeLast()。线程A进行step 1拿到last同时线程B也拿到同样的last；此时由于线程调度上的原因，线程B先执行了step 4删除了最后一个节点，而线程A在此之后才执行step 2， 由于最后一个节点已被删除，线程A这里会报ArrayIndexOutOfBoundsException，而这个错误并不是用户希望看到的。

　　

所以如果要按照类似的方法使用synchronized容器的话还是需要自己加锁。由于这些容器内部的线程安全策略是使用自己的内置锁，所以用户代码加锁的时候需要用到的是容器本身。

 1 public Object getLast(Vector list){
 2 　　synchronized(list){
 3 　　　　int last = list.size() - 1; //1
 4 　　　　return list.get(last); //2
 5 　　}
 6 }
 7 public void removeLast(Vector list){
 8 　　synchronized(list){
 9 　　　　int last = list.size() - 1; //3
10 　　　　list.remove(last); //4
11 　　}
12 }

除了这些用户自定义的复合操作之外，其实iteration也算复合操作，所以也应该加锁。此处应注意两点：

1.  容器自带的Iterator本身不支持并发修改，所以它提供了一个所谓的fail-fast的并发修改报错机制，即容器自身维护一个modCount域，Iterator在创建时记录这个modCount的值，如果在用户遍历容器的过程中modCount值发生了改变，则说明有另一个线程对容器做出了修改，那么Iterator马上会抛出ConcurrentModificationException。

　　　　这个机制严格意义上并不能够100%地探测到并发修改，因为modCount这个域并不是volatile的，在判断　　　　

　　　　if(modCount == expectedModCount)

　　　时也并未加锁。作者描述这个机制是在考虑性能的情况下所做的一个best-effort的努力。总之，不应该对这个机制做过多的依赖。

　      2.  有一些容器自带的方法看起来很无辜，但内部会用到iterator，所以用户用到这些无辜方法的时候还是要加锁。比如我们常用的toString, for-each语法，hashCode, equals, containsAll, removeAll, retainAll, 以其他容器为参数的构造器，等等。而这些方法有时候也是被隐式调用的，很难检查到，比如：

　　　　1 //...add some elements to the set
　　　　2 System.out.println("DEBUG: added ten elements to " + set);

　　　　这里打印时set.toString()方法被隐式调用了。

client locking的问题

由于client code尝试使用容器内部的线程安全机制，所以容易导致starvation和deadlock，这是因为任意代码都可以使用容器的内置锁，散落在各处的线程安全机制使得程序很难维护和debug。如果要解决这个问题，可以把容器克隆到线程内部进行使用，但每次使用的时候都要重新克隆，要考虑克隆本身带来的代价。

Concurrent容器

相比于synchronized容器，Concurrent容器可以提供更高的并发性。
如果需要并发的Map，相比于synchronized Map，可以优先考虑ConcurrentHashMap；同理，相比于synchronized List/Set，可以优先考虑CopyOnWriteArrayList/Set；相比于synchronized SortedMap/Set，可以优先考虑ConcurrentSkipMap/Set。

ConcurrentHashMap

+ 使用了比Hashtable更细粒度的lock striping线程安全策略，支持多个（有限个）线程同时读写。
+ 提供的Iterator是weakly consistent的，容许并发修改。
- size/isEmpty等方法只提供估算值。
- 由于使用的锁对象是private的，不支持client-side locking。（但是提供put-if-absent等复合操作）

CopyOnWriteArrayList

+ 每次改动时创建和发布新的collection copy。
+ 内部array是effectively immutable的，因此发布后可以不加锁地安全访问。
+ 适用于iteration >> modification的情况，如listeners。

BlockingQueue与生产者-消费者

BlockingQueue的最大好处是它不仅是一个简单的容器，它还能提供flow-control，能让程序在消息过多的情况下仍然保持健壮。

特殊的BlockingQueue: SynchronousQueue

一种很特殊的queue，实际上没有内在的存储，只是用于线程间的交接（rendezvous）。适用于消费者够多的情况，比起BlockingQueue的最大好处是没有交接成本。

 1 Thread producer = new Thread("PRODUCER") {
 2 　　public void run() {
 3 　　　　String event = "MY_EVENT";
 4 　　　　try {
 5 　　　　　　queue.put(event); // thread will block here
 6 　　　　　　System.out.printf("[%s] published event : %s %n", Thread.currentThread().getName(), event);
 7 　　　　} catch (InterruptedException e) {
 8 　　　　　　e.printStackTrace();
 9 　　　　}
10 　　}
11 };
12 producer.start(); // starting publisher thread
13 
14 
15 Thread consumer = new Thread("CONSUMER") {
16 　　public void run() {
17 　　　　try {
18 　　　　　　String event = queue.take(); // thread will block here
19 　　　　　　System.out.printf("[%s] consumed event : %s %n", Thread.currentThread().getName(), event);
20 　　　　} catch (InterruptedException e) {
21 　　　　　　e.printStackTrace();
22 　　　　}
23 　　}
24 };
25 consumer.start(); // starting consumer thread
26 
27 [PRODUCER] published event : MY_EVENT
28 [CONSUMER] consumed event : MY_EVENT

 

Synchronizers

所谓的synchronizer，就是能够根据其内部状态调节线程的control flow的对象。

CountDownLatch

主要方法：
　　- countDown
　　- await
CountDownLatch有如一个阀门，在其达到最终状态前阀门关闭，线程不可通过。达到最终状态时，阀门打开，所有线程通过。打开后的阀门永远打开，状态不再改变。

适用情景：

• 等待所依赖的资源全部加载完成后才继续。　
• 初始化顺序中各个service之间的相互等待。 
• 等待所有参与的player都准备好才开始游戏。

FutureTask

主要方法：get
task真正结束前get方法会阻塞，直到task执行结束/被取消/抛异常。

Semaphore

主要方法：
　　- release
　　- acquire

有有限多个permit，acquire时如果permit为0会阻塞，但release可以执行无限多次。

适合：控制可以同时访问某资源的activity数量。可用来实现资源池或将容器设为可以存储有限个元素的容器。

CyclicBarrier

主要方法：await

必须所有线程到达Barrier时，所有线程才能通过。

Latch用来等待事件；Barrier用来等待其它线程。

适用场景：N等N

 1 public class CellularAutomata {
 2     private final Board mainBoard;
 3     private final CyclicBarrier barrier;
 4     private final Worker[] workers;
 5 
 6     public CellularAutomata(Board board) {
 7         this.mainBoard = board;
 8         int count = Runtime.getRuntime().availableProcessors();
 9         this.barrier = new CyclicBarrier(count,
10                 new Runnable() {
11                     public void run() {
12                         mainBoard.commitNewValues();
13                     }});
14         this.workers = new Worker[count];
15         for (int i = 0; i < count; i++)
16             workers[i] = new Worker(mainBoard.getSubBoard(count, i));
17     }
18 
19     private class Worker implements Runnable {
20         private final Board board;
21 
22         public Worker(Board board) { this.board = board; }
23         public void run() {
24             while (!board.hasConverged()) {
25                 for (int x = 0; x < board.getMaxX(); x++)
26                     for (int y = 0; y < board.getMaxY(); y++)
27                         board.setNewValue(x, y, computeValue(x, y));
28                 try {
29                     barrier.await();
30                 } catch (InterruptedException ex) {
31                     return;
32                 } catch (BrokenBarrierException ex) {
33                     return;
34                 }
35             }
36         }
37 
38         private int computeValue(int x, int y) {
39             // Compute the new value that goes in (x,y)
40             return 0;
41         }
42     }
43 
44     public void start() {
45         for (int i = 0; i < workers.length; i++)
46             new Thread(workers[i]).start();
47         mainBoard.waitForConvergence();
48     }
49 }