首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >Java并发编程--Exchanger

Java并发编程--Exchanger

作者头像
在周末
发布2019-08-26 17:56:16
3940
发布2019-08-26 17:56:16
举报
文章被收录于专栏:在周末的专栏在周末的专栏

概述

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

本文参与 腾讯云自媒体分享计划,分享自作者个人站点/博客。
原始发表:2017-11-05 ,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 概述
  • 使用
  • 实现原理
    •   域
      •   方法exchange(V x)
      • 参考资料:
      领券
      问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档