Java并发编程--Exchanger

Posted 2020-10-13 在周末

概述

　　用于线程间数据的交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过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 }

　　方法exchange(V x)

 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并发编程的艺术》