Java併發編程--Exchanger

時間 2019-12-08

標籤 java 併發編程 exchanger 欄目 Java 简体版

原文原文鏈接

概述

　　用於線程間數據的交換。它提供一個同步點，在這個同步點，兩個線程能夠交換彼此的數據。這兩個線程經過exchange方法交換數據，若是第一個線程先執行exchange()方法，它會一直等待第二個線程也執行exchange方法，當兩個線程都到達同步點時，這兩個線程就能夠交換數據，將本線程生產出來的數據傳遞給對方。html

　　Exchanger 可被視爲 SynchronousQueue 的雙向形式。Exchanger在遺傳算法和管道設計等應用中頗有用。java

　　內存一致性：對於經過 Exchanger 成功交換對象的每對線程，每一個線程中在 exchange() 以前的操做 happen-before 從另外一線程中相應的 exchange() 返回的後續操做。node

使用

　　提供的方法：算法

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兩種，看一下這兩個方法的源代碼：app

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