用于线程间数据的交换。它提供一个同步点,在这个同步点,两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据,如果第一个线程先执行exchange()方法,它会一直等待第二个线程也执行exchange方法,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。
Exchanger 可被视为 SynchronousQueue 的双向形式。Exchanger在遗传算法和管道设计等应用中很有用。
内存一致性:对于通过 Exchanger 成功交换对象的每对线程,每个线程中在 exchange() 之前的操作 happen-before 从另一线程中相应的 exchange() 返回的后续操作。
提供的方法:
1 // 等待另一个线程到达此交换点(除非当前线程被中断),然后将给定的对象传送给该线程,并接收该线程的对象。
2 public V exchange(V x) throws InterruptedException
3 //增加超时机制,超过指定时间,抛TimeoutException异常
4 public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
使用示例:
该类使用 Exchanger 在线程间交换缓冲区,因此,在需要时,填充缓冲区的线程获取一个新腾空的缓冲区,并将填满的缓冲区传递给清空缓冲区的线程。
1 class FillAndEmpty {
2 Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
3 DataBuffer initialEmptyBuffer = ... a made-up type
4 DataBuffer initialFullBuffer = ...
5
6 //填充缓冲区线程
7 class FillingLoop implements Runnable {
8 public void run() {
9 DataBuffer currentBuffer = initialEmptyBuffer; //空的缓冲区
10 try {
11 while (currentBuffer != null) {
12 addToBuffer(currentBuffer); //填充数据
13 //如果缓冲区被数据填满,执行exchange。等待清空缓冲区线程也执行exchange方法。当两个线程都到达同步点,交换数据。
14 if (currentBuffer.isFull())
15 currentBuffer = exchanger.exchange(currentBuffer);
16 }
17 } catch (InterruptedException ex) { ... handle ... }
18 }
19 }
20
21 //清空缓冲区线程
22 class EmptyingLoop implements Runnable {
23 public void run() {
24 DataBuffer currentBuffer = initialFullBuffer; //满的缓冲区
25 try {
26 while (currentBuffer != null) {
27 takeFromBuffer(currentBuffer); //清空缓冲区
28 //如果缓冲区被清空,执行exchange。等待填充缓冲区线程也执行exchange方法。当两个线程都到达同步点,交换数据。
29 if (currentBuffer.isEmpty())
30 currentBuffer = exchanger.exchange(currentBuffer);
31 }
32 } catch (InterruptedException ex) { ... handle ...}
33 }
34 }
35
36 void start() {
37 new Thread(new FillingLoop()).start();
38 new Thread(new EmptyingLoop()).start();
39 }
40 }
1 //一个Slot就是一对线程交换数据的地方。使用缓存行填充,提高在小于等于64byte缓存行上隔离性,避免伪共享问题
2 private static final class Slot extends AtomicReference<Object> {
3 // Improve likelihood of isolation on <= 64 byte cache lines
4 long q0, q1, q2, q3, q4, q5, q6, q7, q8, q9, qa, qb, qc, qd, qe;
5 }
6 /*
7 伪共享说明:假设一个类的两个相互独立的属性a和b在内存地址上是连续的(比如FIFO队列的头尾指针),那么它们通常会被加载到相同的cpu cache line里面。并发情况下,如果一个线程修改了a,会导致整个cache line失效(包括b),这时另一个线程来读b,就需要从内存里再次加载了,这种多线程频繁修改ab的情况下,虽然a和b看似独立,但它们会互相干扰,非常影响性能。
8 */
9
10 private static final int CAPACITY = 32;
11
12 //arena最大不会超过FULL,避免空间浪费。如果单核或者双核CPU,FULL=0,只有一个SLot可以用。
13 private static final int FULL = Math.max(0, Math.min(CAPACITY, NCPU / 2) - 1);
14
15 //Slot数组中的元素延迟初始化,等到需要的时候再初始化。声明为volatile变量是因为初始化Slot采用双重检查锁的方式可能带来的问题,见《Java内存模型》。
16 //类似于ConcurrentHashMap的设计思想,减少一些竞争
17 private volatile Slot[] arena = new Slot[CAPACITY];
18
19 //Slot索引的最大值。当线程有太多的CAS竞争时,会递增;当线程自旋等待超时后,会递减。
20 private final AtomicInteger max = new AtomicInteger();
21
22 //自旋等待次数。单核情况下,自旋次数为0;多核情况下为大多数系统线程上下文切换的平均值。该值设置太大会消耗CPU。
23 private static final int SPINS = (NCPU == 1) ? 0 : 2000;
24 //若在超时机制下,自旋次数更少,因为多个检测超时的时间,这是一个经验值。
25 private static final int TIMED_SPINS = SPINS / 20;
每个要进行数据交换的线程在内部会用一个Node来表示。Node类的源码:
1 //每个要进行数据交换的线程在内部会用一个Node来表示。
2 //注意Node继承了AtomicReference,AtomicReference含有域value。其中final域item是用来存储自己要交换出的数据,而域value用来接收其他线程交换来的数据。
3 private static final class Node extends AtomicReference<Object> {
4 /** The element offered by the Thread creating this node. */
5 //要交换的数据
6 public final Object item;
7
8 /** The Thread waiting to be signalled; null until waiting. */
9 //等待唤醒的线程
10 public volatile Thread waiter;
11
12 /**
13 * Creates node with given item and empty hole.
14 * @param item the item
15 */
16 public Node(Object item) {
17 this.item = item;
18 }
19 }
1 //交换数据
2 public V exchange(V x) throws InterruptedException {
3 if (!Thread.interrupted()) {
4 Object v = doExchange((x == null) ? NULL_ITEM : x, false, 0);
5 if (v == NULL_ITEM)
6 return null;
7 if (v != CANCEL)
8 return (V)v;
9 Thread.interrupted(); // Clear interrupt status on IE throw 清除中断标记
10 }
11 throw new InterruptedException();
12 }
13
14 //item为要交换的数据,timed表示是否使用超时机制
15 private Object doExchange(Object item, boolean timed, long nanos) {
16 //创建当前节点me
17 Node me = new Node(item); // Create in case occupying
18 //index是根据当前线程ID计算出的Slot索引
19 int index = hashIndex(); // Index of current slot
20 //CAS失败次数
21 int fails = 0; // Number of CAS failures
22
23 for (;;) {
24 Object y; // Contents of current slot,表示当前Slot可能存在的Node
25 Slot slot = arena[index];
26 //如果slot为null,初始化一个Slot。Slot的延迟初始化采用双重检查锁方式
27 if (slot == null) // Lazily initialize slots
28 createSlot(index); // Continue loop to reread
29 //如果Slot不为空并且该Slot上存在Node,说明该Slot已经被占用。当前线程尝试与该Node交换数据,通过CAS将Slot置为null。
30 else if ((y = slot.get()) != null && // Try to fulfill
31 slot.compareAndSet(y, null)) {
32 Node you = (Node)y; // Transfer item
33 //把要交换的数据item给you,注意此处是赋值给you.value域,you.item域并没有改变,也不能改变(final域),所以下面返回的you.item是you要交换出的数据。
34 //若CAS成功则唤醒you的等待线程返回you的数据item;否则说明you已经和其他线程Node交换数据,当前线程继续下个循环寻找可以交换的Node。
35 if (you.compareAndSet(null, item)) {
36 LockSupport.unpark(you.waiter); //唤醒you的等待线程,因为有可能提前到达的线程被阻塞。
37 return you.item;
38 } // Else cancelled; continue
39 }
40 //如果Slot中无Node,说明该Slot上没有Node与当前线程交换数据,那么当前线程尝试占用该Slot。
41 else if (y == null && // Try to occupy
42 slot.compareAndSet(null, me)) {
43 //如果index为0,则直接调用await等待。index=0说明只有当前线程在等待交换数据,
44 if (index == 0) // Blocking wait for slot 0
45 return timed ?
46 awaitNanos(me, slot, nanos) :
47 await(me, slot);
48
49 //如果index不为0,则自旋等待其他线程前来交换数据,并返回交换后的数据
50 Object v = spinWait(me, slot); // Spin wait for non-0
51 if (v != CANCEL)
52 return v;
53
54 //如果被取消(什么情况下可能被取消?spinWait方法中超过自旋次数或取消),重新建一个Node。继续循环尝试与其他线程交换数据
55 me = new Node(item); // Throw away cancelled node
56 //max递减操作,如果自旋等待超时,可能是因为需要交换的线程数少,而Slot的数量过多导致。尝试减少Slot的数量,增加两个线程落到同一个Slot的几率。
57 int m = max.get();
58 //如果当前的最大索引值max过大
59 if (m > (index >>>= 1)) // Decrease index //减小索引,index右移一位。如果一直自旋等不到其他线程,会一直右移直到index变为0阻塞线程。
60 max.compareAndSet(m, m - 1); // Maybe shrink table
61 }
62 //如果在第一个Slot上竞争失败2次,说明该Slot竞争激烈。index递减,换个Slot继续
63 else if (++fails > 1) { // Allow 2 fails on 1st slot
64 int m = max.get();
65 //如果累计竞争失败超过3次,说明多个Slot竞争太多。尝试递增max。并将index置为m+1,换个Slot,继续循环。
66 if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
67 index = m + 1; // Grow on 3rd failed slot
68 else if (--index < 0) //index递减,继续循环。
69 index = m; // Circularly traverse 防止越界
70 }
71 }
72 }
首先判断Slot是否为null,如果为null则在当前index上创建一个Slot,创建Slot时为了提供效率使用了双重检查锁定模型,看一下createSlot()代码。
1 //在指定的index位置创建Slot
2 private void createSlot(int index) {
3 // Create slot outside of lock to narrow sync region
4 Slot newSlot = new Slot();
5 Slot[] a = arena;
6 synchronized (a) {
7 if (a[index] == null) //第二次检查
8 a[index] = newSlot;
9 }
10 }
如果Slot上存在Node,则尝试与该Node交换数据。若交换成功,则唤醒被阻塞的线程;如果交换失败,继续循环。
如果Slot上没有Node,说明是当前线程先到达Slot。当index=0时,当前线程阻塞等待另一个线程前来交换数据;当index!=0时,当前线程自旋等待其他线程前来交换数据。其中阻塞等待有await和awaitNanos两种,看一下这两个方法的源代码:
1 //在index为0的slot上等待其他线程前来交换数据。先尝试自旋等待,如果自旋期间未成功,则进入阻塞当前线程。除非当前线程被中断。
2 private static Object await(Node node, Slot slot) {
3 Thread w = Thread.currentThread();
4 int spins = SPINS;
5 for (;;) {
6 Object v = node.get();
7 //如果已经被其他线程填充了值,则返回这个值
8 if (v != null)
9 return v;
10 else if (spins > 0) // Spin-wait phase
11 --spins; //先自旋等待spins次,
12 else if (node.waiter == null) // Set up to block next
13 node.waiter = w;
14 else if (w.isInterrupted()) // Abort on interrupt 如果当前线程被中断,尝试取消该Node。并
15 tryCancel(node, slot);
16 else // Block
17 LockSupport.park(node); //阻塞当前线程
18 }
19 }
20
21 //增加超时机制,其他同await方法。
22 private Object awaitNanos(Node node, Slot slot, long nanos) {
23 int spins = TIMED_SPINS;
24 long lastTime = 0;
25 Thread w = null;
26 for (;;) {
27 Object v = node.get();
28 if (v != null)
29 return v;
30 long now = System.nanoTime();
31 if (w == null)
32 w = Thread.currentThread();
33 else
34 nanos -= now - lastTime;
35 lastTime = now;
36 if (nanos > 0) {
37 if (spins > 0)
38 --spins;
39 else if (node.waiter == null)
40 node.waiter = w;
41 else if (w.isInterrupted())
42 tryCancel(node, slot);
43 else
44 LockSupport.parkNanos(node, nanos);
45 }
46 else if (tryCancel(node, slot) && !w.isInterrupted())
47 //超时后,如果当前线程没有被中断,那么从Slot数组的其他位置看看有没有等待交换数据的节点
48 return scanOnTimeout(node);
49 }
50 }
51
52 //尝试取消节点
53 private static boolean tryCancel(Node node, Slot slot) {
54 //将node的value域CAS置为CANCEL。如果失败,表示已经被取消。
55 if (!node.compareAndSet(null, CANCEL))
56 return false;
57 //将当前slot置空
58 if (slot.get() == node) // pre-check to minimize contention
59 slot.compareAndSet(node, null);
60 return true;
61 }
62
63 //该方法仅在index为0的slot上等待重试时被调用。
64 //超时后,如果当前线程没有被中断,那么从Slot数组的其他位置看看有没有等待交换数据的节点。如果有就交换。
65 //当该线程因为超时而被取消时,有可能以前进入的线程仍然在等待(即存在等待交换的Node,只是没有在当前Slot 0内)。所以遍历全部的Slot找可能存在的Node。
66 private Object scanOnTimeout(Node node) {
67 Object y;
68 for (int j = arena.length - 1; j >= 0; --j) {
69 Slot slot = arena[j];
70 if (slot != null) {
71 while ((y = slot.get()) != null) {
72 if (slot.compareAndSet(y, null)) {
73 Node you = (Node)y;
74 if (you.compareAndSet(null, node.item)) {
75 LockSupport.unpark(you.waiter);
76 return you.item;
77 }
78 }
79 }
80 }
81 }
82 //没找到则返回CANCEL
83 return CANCEL;
84 }
再看一下当index!=0时,当前线程自旋等待其他线程的实现,spinWait方法。如果超过自旋次数,则取消Node,然后重新建一个Node,减小index(index >>>= 1)且有可能减小max(max.compareAndSet(m, m - 1)),继续循环。如果如果经过多次自旋等待还是没有其他线程交换数据,index会一直右移直到变为0,当index=0时,就会阻塞等待其他线程交换数据。
//在index不为0的slot中,自旋等待。
private static Object spinWait(Node node, Slot slot) {
int spins = SPINS;
for (;;) {
Object v = node.get();
if (v != null)
return v;
else if (spins > 0)
--spins;
else
//超过自旋次数,取消当前Node
tryCancel(node, slot);
}
}
如果走到最后一个分支(++fails > ),说明竞争激烈。如果在第一个Slot上竞争失败2次,说明该Slot竞争激烈,index递减,换个Slot继续;如果累计竞争失败超过3次,说明存在多个Slot竞争激烈,尝试递增max,增加Slot的个数,并将index置为m+1,换个Slot继续。
(《java.util.concurrent 包源码阅读》18 Exchanger)http://www.cnblogs.com/wanly3643/p/3939552.html
(Jdk1.6 JUC源码解析(27)-Exchanger)http://brokendreams.iteye.com/blog/2253956
(Java多线程 -- JUC包源码分析16 -- Exchanger源码分析)http://blog.csdn.net/chunlongyu/article/details/52504895
《Java并发编程的艺术》