[JCIP笔记](五)JDK并发包
这一节来讲一讲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也算复合操作,所以也应该加锁。此处应注意两点:
- 容器自带的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 }