并发编程之Exchanger原理与使用

 前言

网站的建设成都创新互联专注网站定制,经验丰富,不做模板,主营网站定制开发.小程序定制开发,H5页面制作!给你焕然一新的设计体验!已为成都展览展示等企业提供专业服务。

在JUC包中，除了一些常用的或者说常见的并发工具类(ReentrantLock，CountDownLatch，CyclicBarrier，Semaphore)等，还有一个不常用的线程同步器类 —— Exchanger。

Exchanger是适用在两个线程之间数据交换的并发工具类，它的作用是找到一个同步点，当两个线程都执行到了同步点(exchange方法)之后(有一个没有执行到就一直等待，也可以设置等待超时时间)，就将自身线程的数据与对方交换。

Exchanger 是什么?

它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这个两个线程通过exchange方法交换数据，如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。因此使用Exchanger的中断时成对的线程使用exchange()方法，当有一对线程到达了同步点，就会进行交换数据，因此该工具类的线程对象是成对的。

线程可以在成对内配对和交换元素的同步点。每个线程在输入exchange方法时提供一些对象，与合作者线程匹配，并在返回时接收其合作伙伴的对象。交换器可以被视为一个的双向形式的SynchroniuzedQueue。交换器在诸如遗传算法和管道设计的应用中可能是有用的。

一个用于两个工作线程之间交换数据的封装工具类，简单说就是一个线程在完成一定事务后想与另一个线程交换数据，则第一个先拿出数据的线程会一直等待第二个线程，直到第二个线程拿着数据到来时才能彼此交换对应数据。

Exchanger 用法

  • Exchanger 泛型类型，其中V表示可交换的数据类型
  • V exchanger(V v)：等待另一个线程到达此交换点(除非当前线程被中断)，然后将给定的对象传送该线程，并接收该线程的对象。
  • V exchanger(V v, long timeout, TimeUnit unit)：等待另一个线程到达此交换点(除非当前线程被中断或超出类指定的等待时间)，然后将给定的对象传送给该线程，并接收该线程的对象。

应用场景

Exchanger可以用于遗传算法，遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出2个交配结果。

Exchanger也可以用于校对工作。比如我们需要将纸制银流通过人工的方式录入成电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对这两个Excel数据进行校对，看看是否录入的一致

Exchanger的典型应用场景是：一个任务在创建对象，而这些对象的生产代价很高，另一个任务在消费这些对象。通过这种方式，可以有更多的对象在被创建的同时被消费。

案例说明

Exchanger 用于两个线程间交换数据，当然实际参与的线程可以不止两个，测试用例如下：

 
 
 
 
    
  
  
  
  
  1. private static void test1() throws InterruptedException {
    2. 
  
  
  
  
  2.         Exchanger exchanger = new Exchanger<>();
    3. 
  
  
  
  
  3.         CountDownLatch countDownLatch = new CountDownLatch(5);
    4. 
  
  
  
  
  4. 
  
  
  
  
  5.         for (int i = 0; i < 5; i++) {
    6. 
  
  
  
  
  6.             new Thread(() ->  {
    7. 
  
  
  
  
  7. 
  
  
  
  
  8.                 try {
    9. 
  
  
  
  
  9.                     String origMsg = RandomStringUtils.randomNumeric(6);
    10. 
  
  
  
  
  10. 
  
  
  
  
  11.                     // 先到达的线程会在此等待，直到有一个线程跟它交换数据或者等待超时
    12. 
  
  
  
  
  12.                     String exchangeMsg = exchanger.exchange(origMsg,5, TimeUnit.SECONDS);
    13. 
  
  
  
  
  13. 
  
  
  
  
  14.                     System.out.println(Thread.currentThread().getName() + "\t origMsg:" + origMsg + "\t exchangeMsg:" + exchangeMsg);
    15. 
  
  
  
  
  15.                 } catch (InterruptedException e) {
    16. 
  
  
  
  
  16.                     e.printStackTrace();
    17. 
  
  
  
  
  17.                 } catch (TimeoutException e) {
    18. 
  
  
  
  
  18.                     e.printStackTrace();
    19. 
  
  
  
  
  19.                 }finally {
    20. 
  
  
  
  
  20.                     countDownLatch.countDown();
    21. 
  
  
  
  
  21.                 }
    22. 
  
  
  
  
  22. 
  
  
  
  
  23.             },String.valueOf(i)).start();
    24. 
  
  
  
  
  24.         }
    25. 
  
  
  
  
  25. 
  
  
  
  
  26.         countDownLatch.await();
    27. 
  
  
  
  
  27.     }
    28.

 第5个线程因为没有匹配的线程而等待超时，输出如下：

 
 
 
 
    
  
  
  
  
  1. 0  origMsg:524053  exchangeMsg:098544
    2. 
  
  
  
  
  2. 3  origMsg:433246  exchangeMsg:956604
    3. 
  
  
  
  
  3. 4  origMsg:098544  exchangeMsg:524053
    4. 
  
  
  
  
  4. 1  origMsg:956604  exchangeMsg:433246
    5. 
  
  
  
  
  5. java.util.concurrent.TimeoutException
    6. 
  
  
  
  
  6.  at java.util.concurrent.Exchanger.exchange(Exchanger.java:626)
    7. 
  
  
  
  
  7.  at com.nuih.juc.ExchangerDemo.lambda$test1$0(ExchangerDemo.java:37)
    8. 
  
  
  
  
  8.  at java.lang.Thread.run(Thread.java:748)
    9.

 上述测试用例是比较简单，可以模拟消息消费的场景来观察Exchanger的行为，测试用例如下：

 
 
 
 
    
  
  
  
  
  1. private static void test2() throws InterruptedException {
    2. 
  
  
  
  
  2.         Exchanger exchanger = new Exchanger<>();
    3. 
  
  
  
  
  3.         CountDownLatch countDownLatch = new CountDownLatch(4);
    4. 
  
  
  
  
  4.         CyclicBarrier cyclicBarrier = new CyclicBarrier(4);
    5. 
  
  
  
  
  5. 
  
  
  
  
  6.         // 生产者
    7. 
  
  
  
  
  7.         Runnable producer = new Runnable() {
    8. 
  
  
  
  
  8.             @Override
    9. 
  
  
  
  
  9.             public void run() {
    10. 
  
  
  
  
  10.                 try{
    11. 
  
  
  
  
  11.                     cyclicBarrier.await();
    12. 
  
  
  
  
  12. 
  
  
  
  
  13.                     for (int i = 0; i < 5; i++) {
    14. 
  
  
  
  
  14.                         String msg = RandomStringUtils.randomNumeric(6);
    15. 
  
  
  
  
  15.                         exchanger.exchange(msg,5,TimeUnit.SECONDS);
    16. 
  
  
  
  
  16.                         System.out.println(Thread.currentThread().getName() + "\t producer msg -> " + msg + " ,\t i -> " + i);
    17. 
  
  
  
  
  17.                     }
    18. 
  
  
  
  
  18. 
  
  
  
  
  19.                 }catch (Exception e){
    20. 
  
  
  
  
  20.                     e.printStackTrace();
    21. 
  
  
  
  
  21.                 }finally {
    22. 
  
  
  
  
  22.                     countDownLatch.countDown();
    23. 
  
  
  
  
  23.                 }
    24. 
  
  
  
  
  24.             }
    25. 
  
  
  
  
  25.         };
    26. 
  
  
  
  
  26. 
  
  
  
  
  27.         // 消费者
    28. 
  
  
  
  
  28.         Runnable consumer = new Runnable() {
    29. 
  
  
  
  
  29.             @Override
    30. 
  
  
  
  
  30.             public void run() {
    31. 
  
  
  
  
  31.                 try{
    32. 
  
  
  
  
  32.                     cyclicBarrier.await();
    33. 
  
  
  
  
  33.                     for (int i = 0; i < 5; i++) {
    34. 
  
  
  
  
  34.                         String msg = exchanger.exchange(null,5,TimeUnit.SECONDS);
    35. 
  
  
  
  
  35.                         System.out.println(Thread.currentThread().getName() + "\t consumer msg -> " + msg + ",\t" + i);
    36. 
  
  
  
  
  36.                     }
    37. 
  
  
  
  
  37. 
  
  
  
  
  38.                 }catch (Exception e){
    39. 
  
  
  
  
  39.                     e.printStackTrace();
    40. 
  
  
  
  
  40.                 }finally {
    41. 
  
  
  
  
  41.                     countDownLatch.countDown();
    42. 
  
  
  
  
  42.                 }
    43. 
  
  
  
  
  43.             }
    44. 
  
  
  
  
  44.         };
    45. 
  
  
  
  
  45. 
  
  
  
  
  46.         for (int i = 0; i < 2; i++){
    47. 
  
  
  
  
  47.             new Thread(producer).start();
    48. 
  
  
  
  
  48.             new Thread(consumer).start();
    49. 
  
  
  
  
  49.         }
    50. 
  
  
  
  
  50. 
  
  
  
  
  51.         countDownLatch.await();
    52. 
  
  
  
  
  52.     }
    53.

 输出如下，上面生产者和消费者线程数是一样的，循环次数也是一样的，但是还是出现等待超时的情形：

 
 
 
 
    
  
  
  
  
  1. Thread-3  consumer msg -> null, 0
    2. 
  
  
  
  
  2. Thread-1  consumer msg -> null, 0
    3. 
  
  
  
  
  3. Thread-1  consumer msg -> null, 1
    4. 
  
  
  
  
  4. Thread-2  producer msg -> 640010 ,  i -> 0
    5. 
  
  
  
  
  5. Thread-2  producer msg -> 733133 ,  i -> 1
    6. 
  
  
  
  
  6. Thread-3  consumer msg -> null, 1
    7. 
  
  
  
  
  7. Thread-3  consumer msg -> 476520, 2
    8. 
  
  
  
  
  8. Thread-1  consumer msg -> 640010, 2
    9. 
  
  
  
  
  9. Thread-1  consumer msg -> null, 3
    10. 
  
  
  
  
  10. Thread-0  producer msg -> 993414 ,  i -> 0
    11. 
  
  
  
  
  11. Thread-0  producer msg -> 292745 ,  i -> 1
    12. 
  
  
  
  
  12. Thread-2  producer msg -> 476520 ,  i -> 2
    13. 
  
  
  
  
  13. Thread-2  producer msg -> 408446 ,  i -> 3
    14. 
  
  
  
  
  14. Thread-3  consumer msg -> null, 3
    15. 
  
  
  
  
  15. Thread-1  consumer msg -> 292745, 4
    16. 
  
  
  
  
  16. Thread-2  producer msg -> 251971 ,  i -> 4
    17. 
  
  
  
  
  17. Thread-0  producer msg -> 078939 ,  i -> 2
    18. 
  
  
  
  
  18. Thread-3  consumer msg -> 251971, 4
    19. 
  
  
  
  
  19. java.util.concurrent.TimeoutException
    20. 
  
  
  
  
  20.  at java.util.concurrent.Exchanger.exchange(Exchanger.java:626)
    21. 
  
  
  
  
  21.  at com.nuih.juc.ExchangerDemo$1.run(ExchangerDemo.java:70)
    22. 
  
  
  
  
  22.  at java.lang.Thread.run(Thread.java:748)
    23. 
  
  
  
  
  23. 
  
  
  
  
  24. Process finished with exit code 0
    25.

 这种等待超时是概率出现的，这是为啥?

因为系统调度的不均衡和Exchanger底层的大量自旋等待导致这4个线程并不是调用exchanger成功的次数并不一致。另外从输出可以看出，消费者线程并没有像我们想的那样跟生产者线程一一匹配，生产者线程有时也充当来消费者线程，这是为啥?因为Exchanger匹配时完全不关注这个线程的角色，两个线程之间的匹配完全由调度决定的，即CPU同时执行来或者紧挨着执行来两个线程，这两个线程就匹配成功来。

源码分析

Exchanger 类图

其内部主要变量和方法如下：

成员属性

 
 
 
 
    
  
  
  
  
  1. // ThreadLocal变量，每个线程都有之间的一个副本
    2. 
  
  
  
  
  2. private final Participant participant;
    3. 
  
  
  
  
  3. // 高并发下使用的，保存待匹配的Node实例
    4. 
  
  
  
  
  4. private volatile Node[] arena;
    5. 
  
  
  
  
  5. // 低并发下，arena未初始化时使用的保存待匹配的Node实例
    6. 
  
  
  
  
  6. private volatile Node slot;
    7. 
  
  
  
  
  7. // 初始值为0，当创建arena后被负责SEQ，用来记录arena数组的可用最大索引，
    8. 
  
  
  
  
  8. // 会随着并发的增大而增大直到等于最大值FULL，
    9. 
  
  
  
  
  9. // 会随着并行的线程逐一匹配成功而减少恢复成初始值
    10. 
  
  
  
  
  10. private volatile int bound;
    11.

 还有多个表示字段偏移量的静态属性，通过static代码块初始化，如下：

 
 
 
 
    
  
  
  
  
  1. // Unsafe mechanics
    2. 
  
  
  
  
  2. private static final sun.misc.Unsafe U;
    3. 
  
  
  
  
  3. private static final long BOUND;
    4. 
  
  
  
  
  4. private static final long SLOT;
    5. 
  
  
  
  
  5. private static final long MATCH;
    6. 
  
  
  
  
  6. private static final long BLOCKER;
    7. 
  
  
  
  
  7. private static final int ABASE;
    8. 
  
  
  
  
  8. static {
    9. 
  
  
  
  
  9.     int s;
    10. 
  
  
  
  
  10.     try {
    11. 
  
  
  
  
  11.         U = sun.misc.Unsafe.getUnsafe();
    12. 
  
  
  
  
  12.         Class ek = Exchanger.class;
    13. 
  
  
  
  
  13.         Class nk = Node.class;
    14. 
  
  
  
  
  14.         Class ak = Node[].class;
    15. 
  
  
  
  
  15.         Class tk = Thread.class;
    16. 
  
  
  
  
  16.         BOUND = U.objectFieldOffset
    17. 
  
  
  
  
  17.             (ek.getDeclaredField("bound"));
    18. 
  
  
  
  
  18.         SLOT = U.objectFieldOffset
    19. 
  
  
  
  
  19.             (ek.getDeclaredField("slot"));
    20. 
  
  
  
  
  20.         MATCH = U.objectFieldOffset
    21. 
  
  
  
  
  21.             (nk.getDeclaredField("match"));
    22. 
  
  
  
  
  22.         BLOCKER = U.objectFieldOffset
    23. 
  
  
  
  
  23.             (tk.getDeclaredField("parkBlocker"));
    24. 
  
  
  
  
  24.         s = U.arrayIndexScale(ak);
    25. 
  
  
  
  
  25.         // ABASE absorbs padding in front of element 0
    26. 
  
  
  
  
  26.         ABASE = U.arrayBaseOffset(ak) + (1 << ASHIFT);
    27. 
  
  
  
  
  27. 
  
  
  
  
  28.     } catch (Exception e) {
    29. 
  
  
  
  
  29.         throw new Error(e);
    30. 
  
  
  
  
  30.     }
    31. 
  
  
  
  
  31.     if ((s & (s-1)) != 0 || s > (1 << ASHIFT))
    32. 
  
  
  
  
  32.         throw new Error("Unsupported array scale");
    33. 
  
  
  
  
  33. }
    34.

 Exchanger 定义来多个静态变量，如下：

 
 
 
 
    
  
  
  
  
  1. // 初始化arena时使用， 1 << ASHIFT 是一个缓存行的大小，避免来不同的Node落入到同一个高速缓存行
    2. 
  
  
  
  
  2. // 这里实际是把数组容量扩大来8倍，原来索引相邻的两个元素，扩容后中间隔来7个元素，从元素的起始地址上看就隔来8个元素，中间的7个都是空的，为来避免原来相邻的两个元素都落入到同一个缓存行中
    3. 
  
  
  
  
  3. // 因为arena是对象数组，一个元素占8字节，8个就是64字节
    4. 
  
  
  
  
  4. private static final int ASHIFT = 7;
    5. 
  
  
  
  
  5. // arena 数组元素的索引最大值即255
    6. 
  
  
  
  
  6. private static final int MMASK = 0xff;
    7. 
  
  
  
  
  7. // arena 数组的最大长度即256
    8. 
  
  
  
  
  8. private static final int SEQ = MMASK + 1;
    9. 
  
  
  
  
  9. // 获取CPU核数
    10. 
  
  
  
  
  10. private static final int NCPU = Runtime.getRuntime().availableProcessors();
    11. 
  
  
  
  
  11. // 实际的数组长度，因为是线程两两配对的，所以最大长度是核数除以2
    12. 
  
  
  
  
  12. static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;
    13. 
  
  
  
  
  13. // 自旋等待的次数
    14. 
  
  
  
  
  14. private static final int SPINS = 1 << 10;
    15. 
  
  
  
  
  15. // 如果交换的对象是null，则返回此对象
    16. 
  
  
  
  
  16. private static final Object NULL_ITEM = new Object();
    17. 
  
  
  
  
  17. // 如果等待超时导致交换失败，则返回此对象
    18. 
  
  
  
  
  18. private static final Object TIMED_OUT = new Object();
    19.

 内部类

Exchanger类中有两个内部类，一个Node，一个Participant。 Participant继承了ThreadLocal并且重写了其initialValue方法，返回一个Node对象。其定义如下：

 
 
 
 
    
  
  
  
  
  1. @sun.misc.Contended static final class Node {
    2. 
  
  
  
  
  2.     int index;              // Arena index
    3. 
  
  
  
  
  3.     int bound;              // Last recorded value of Exchanger.bound
    4. 
  
  
  
  
  4.     int collides;           // Number of CAS failures at current bound
    5. 
  
  
  
  
  5.     int hash;               // Pseudo-random for spins
    6. 
  
  
  
  
  6.     Object item;            // This thread's current item
    7. 
  
  
  
  
  7.     volatile Object match;  // Item provided by releasing thread
    8. 
  
  
  
  
  8.     volatile Thread parked; // Set to this thread when parked, else null
    9. 
  
  
  
  
  9. }
    10. 
  
  
  
  
  10. 
  
  
  
  
  11. /** The corresponding thread local class */
    12. 
  
  
  
  
  12. static final class Participant extends ThreadLocal {
    13. 
  
  
  
  
  13.     public Node initialValue() { return new Node(); }
    14. 
  
  
  
  
  14. }
    15.

 其中Contended注解是为了避免高速缓存行导致的伪共享问题

  • index用来记录arena数组的索引
  • bound用于记录上一次的Exchanger bound属性
  • collides用于记录在bound不变的情况下CAS抢占失败的次数
  • hash是自旋等待时计算随机数使用的
  • item表示当前线程请求交换的对象
  • match是同其它线程交换的结果，match不为null表示交换成功
  • parked为跟该Node关联的处于休眠状态的线程。

重要方法

exchange()方法

 
 
 
 
    
  
  
  
  
  1. @SuppressWarnings("unchecked")
    2. 
  
  
  
  
  2. public V exchange(V x) throws InterruptedException {
    3. 
  
  
  
  
  3.     Object v;
    4. 
  
  
  
  
  4.     Object item = (x == null) ? NULL_ITEM : x; // translate null args
    5. 
  
  
  
  
  5.     if ((arena != null || // 是null就执行后面的方法
    6. 
  
  
  
  
  6.          (v = slotExchange(item, false, 0L)) == null) &&
    7. 
  
  
  
  
  7.         // 如果执行slotExchange有结果就执行后面的，否则返回
    8. 
  
  
  
  
  8.         ((Thread.interrupted() || // 非中断则执行后面的方法
    9. 
  
  
  
  
  9.           (v = arenaExchange(item, false, 0L)) == null)))
    10. 
  
  
  
  
  10.         throw new InterruptedException();
    11. 
  
  
  
  
  11.     return (v == NULL_ITEM) ? null : (V)v;
    12. 
  
  
  
  
  12. }
    13.

 exchange 方法的执行步骤：

  1. 如果执行 soltExchange 有结果就执行后面的 arenaExchange;
  2. 如果 slot 被占用，就执行 arenaExchange;
  3. 返回的数据 v 是对方线程的数据项;
  4. 总结即：如果A线程先调用，那么A的数据项存储的 item中，则B线程的数据项存储在 math 中;
  5. 当没有多线程并发操作 Exchange 的时候，使用 slotExchange 就足够了，slot 是一个 node 对象;
  6. 当出现并发了，一个 slot 就不够了，就需要使用一个 node 数组 arena 操作了。

slotExchange()方法

slotExchange 是基于slot属性来完成交换的，调用soltExchange方法时，如果slot属性为null，当前线程会将slot属性由null修改成当前线程的Node，如果修改失败则下一次for循环走solt属性不为null的逻辑，如果修改成功则自旋等待，自旋一定次数后通过Unsafe的park方法当当前线程休眠，可以指定休眠的时间，如果没有指定则无限期休眠直到被唤醒;无论是因为线程中断被唤醒，等待超时被唤醒还是其它线程unpark唤醒的，都会检查当前线程的Node的属性释放为null，如果不为null说明交互成功，返回该对象;否则返回null或者TIME_OUT,在返回前会将item，match等属性置为null，保存之前自旋时计算的hash值，方便下一次调用slotExchange。

调用slotExchange方法时，如果slot属性不为null，则当前线程会尝试将其修改null，如果cas修改成功，表示当前线程与slot属性对应的线程匹配成功，会获取slot属性对应Node的item属性，将当前线程交换的对象保存到slot属性对应的Node的match属性，然后唤醒获取slot属性对应Node的waiter属性，即处理休眠状态的线程，至此交换完成，同样的在返回前需要将item，match等属性置为null，保存之前自旋时计算的hash置，方便下一次调用slotExchange;如果cas修改slot属性失败，说明有其它线程也在抢占slot，则初始化arena属性，下一次for循环因为arena属性不为null，直接返回null，从而通过arenaExchange完成交换。

 
 
 
 
    
  
  
  
  
  1. // arena 为null是会调用此方法，返回null表示交换失败
    2. 
  
  
  
  
  2. // item是交换的对象，timed表示是否等待指定的时间，为false表示无限期等待，ns为等待时间
    3. 
  
  
  
  
  3. private final Object slotExchange(Object item, boolean timed, long ns) {
    4. 
  
  
  
  
  4.     // 获取当前线程关联的participant Node
    5. 
  
  
  
  
  5.     Node p = participant.get();
    6. 
  
  
  
  
  6.     Thread t = Thread.currentThread();
    7. 
  
  
  
  
  7.     // 被中断，返回null
    8. 
  
  
  
  
  8.     if (t.isInterrupted()) // preserve interrupt status so caller can recheck
    9. 
  
  
  
  
  9.         return null;
    10. 
  
  
  
  
  10.     
    11. 
  
  
  
  
  11.     for (Node q;;) {
    12. 
  
  
  
  
  12.         if ((q = slot) != null) { // slot 不为null
    13. 
  
  
  
  
  13.             // 将slot置为null，slot对应的线程与当前线程匹配成功
    14. 
  
  
  
  
  14.             if (U.compareAndSwapObject(this, SLOT, q, null)) {
    15. 
  
  
  
  
  15.                 Object v = q.item;
    16. 
  
  
  
  
  16.                 // 保存item，即完成交互
    17. 
  
  
  
  
  17.                 q.match = item;
    18. 
  
  
  
  
  18.                 // 唤醒q对应的处于休眠状态的线程
    19. 
  
  
  
  
  19.                 Thread w = q.parked;
    20. 
  
  
  
  
  20.                 if (w != null)
    21. 
  
  
  
  
  21.                     U.unpark(w);
    22. 
  
  
  
  
  22.                 return v;
    23. 
  
  
  
  
  23.             }
    24. 
  
  
  
  
  24.             // slot修改失败，其它某个线程抢占来该slot，多个线程同时调用exchange方法会触发此逻辑
    25. 
  
  
  
  
  25.             // bound等于0表示未初始化，此处校验避免重复初始化
    26. 
  
  
  
  
  26.             if (NCPU > 1 && bound == 0 &&
    27. 
  
  
  
  
  27.                 U.compareAndSwapInt(this, BOUND, 0, SEQ))
    28. 
  
  
  
  
  28.                 arena = new Node[(FULL + 2) << ASHIFT];
    29. 
  
  
  
  
  29.         }
    30. 
  
  
  
  
  30.         else if (arena != null)
    31. 
  
  
  
  
  31.             return null; // carena不为null，通过arenaExchange交互
    32. 
  
  
  
  
  32.         else {
    33. 
  
  
  
  
  33.             // slot和arena都为null
    34. 
  
  
  
  
  34.             p.item = item;
    35. 
  
  
  
  
  35.             // 修改slot为p，修改成功则终止循环
    36. 
  
  
  
  
  36.             if (U.compareAndSwapObject(this, SLOT, null, p))
    37. 
  
  
  
  
  37.                 break;
    38. 
  
  
  
  
  38.             // 修改失败则继续for循环，将otem恢复成null
    39. 
  
  
  
  
  39.             p.item = null;
    40. 
  
  
  
  
  40.         }
    41. 
  
  
  
  
  41.     }
    42. 
  
  
  
  
  42. 
  
  
  
  
  43.     // 将slot修改为p后会进入此分支
    44. 
  
  
  
  
  44.     int h = p.hash; // hash初始为0
    45. 
  
  
  
  
  45.     long end = timed ? System.nanoTime() + ns : 0L;
    46. 
  
  
  
  
  46.     int spins = (NCPU > 1) ? SPINS : 1;
    47. 
  
  
  
  
  47.     Object v;
    48. 
  
  
  
  
  48.     // match保存着同其他线程交换的对象，如果不为null，说明交换成功了
    49. 
  
  
  
  
  49.     while ((v = p.match) == null) {
    50. 
  
  
  
  
  50.         // 执行自旋等待
    51. 
  
  
  
  
  51.         if (spins > 0) {
    52. 
  
  
  
  
  52.         h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
    53. 
  
  
  
  
  53.         if (h == 0)
    54. 
  
  
  
  
  54.             h = SPINS | (int)t.getId(); 初始化h
    55. 
  
  
  
  
  55.         // 只有生成的h小于0时才减少spins
    56. 
  
  
  
  
  56.         else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
    57. 
  
  
  
  
  57.             Thread.yield();
    58. 
  
  
  
  
  58.         }
    59. 
  
  
  
  
  59.         // slot被修改了，已经有匹配的线程，重新自旋，读取属性，因为是先修改slot再修改属性的，两者因为CPU调度的问题可能有时间差
    60. 
  
  
  
  
  60.         else if (slot != p)
    61. 
  
  
  
  
  61.             spins = SPINS;
    62. 
  
  
  
  
  62.         // 线程没有被中断且arena为null
    63. 
  
  
  
  
  63.         else if (!t.isInterrupted() && arena == null &&
    64. 
  
  
  
  
  64.                  (!timed || (ns = end - System.nanoTime()) > 0L)) {
    65. 
  
  
  
  
  65.             U.putObject(t, BLOCKER, this);
    66. 
  
  
  
  
  66.             p.parked = t;
    67. 
  
  
  
  
  67.             if (slot == p)
    68. 
  
  
  
  
  68.                 U.park(false, ns);
    69. 
  
  
  
  
  69.             // 线程被唤醒，继续下一次for循环
    70. 
  
  
  
  
  70.             // 如果是因为等待超时而被唤醒，下次for循环进入下没的else if分支，返回TIMED_OUT
    71. 
  
  
  
  
  71.             p.parked = null;
    72. 
  
  
  
  
  72.                 U.putObject(t, BLOCKER, null);
    73. 
  
  
  
  
  73.         }
    74. 
  
  
  
  
  74.         // 将slot修改成p
    75. 
  
  
  
  
  75.         else if (U.compareAndSwapObject(this, SLOT, p, null)) {
    76. 
  
  
  
  
  76.             // timed为flase，无限期等待，因为中断被唤醒返回null
    77. 
  
  
  
  
  77.             // timed为ture，因为超时被唤醒，返回TIMED_OUT，因为中断被唤醒返回null
    78. 
  
  
  
  
  78.             v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
    79. 
  
  
  
  
  79.             break;
    80. 
  
  
  
  
  80.         }
    81. 
  
  
  
  
  81.     }
    82. 
  
  
  
  
  82.     // 修改match为null，item为null，保存h，下一次exchange是h就不是初始值为0了
    83. 
  
  
  
  
  83.     U.putOrderedObject(p, MATCH, null);
    84. 
  
  
  
  
  84.     // 重置 item
    85. 
  
  
  
  
  85.     p.item = null;
    86. 
  
  
  
  
  86.     // 保留伪随机数,供下次种子数字
    87. 
  
  
  
  
  87.     p.hash = h;
    88. 
  
  
  
  
  88.     // 返回
    89. 
  
  
  
  
  89.     return v;
    90. 
  
  
  
  
  90. }
    91.

 总结一下上面执行的逻辑：

  • Exchange 使用了对象池的技术，将对象保存在 ThreadLocal 中，这个对象(Node)封装了数据项，线程对象等关键数据;
  • 第一个线程进入的时候，会将数据放到池化对象中，并赋值给 slot 的 item，并阻塞自己(通常不会立即阻塞，而是使用 yield 自旋一会儿)，等待对方取值;
  • 当第二个线程进入的时候，会拿出存储在 slot item 中的值，然后对 slot 的 match 赋值，并唤醒上次阻塞的线程;
  • 当第一个线程阻塞被唤醒后，说明对方取到值了，就获取 slot 的 match 值，并重置 slot 的数据和池化对象的数据，并返回自己的数据;
  • 如果超时了，就返回 Time_out 对象;
  • 如果线程中断了，就返回 null。

在该方法中，会返回 2 种结果，一是有效的 item，二是 null 要么是线程竞争使用 slot 了，创建了 arena 数组，要么是线程中断了。

通过一副图来看看具体逻辑

arenaExchange() 方法

arenaExchange是基于arena属性完成交换的，整体逻辑比较复杂，有以下几个要点：

  • m的初始值就是0，index的初始值也是0，两个都是大于等于0且i不大于m，当某个线程多次尝试抢占index对应数组元素的Node都失败的情形下则尝试将m加1，然后抢占m加1对应的新数组元素，将其由null修改成当前线程关联的Node，然后自旋等待匹配;如果自旋结束，没有匹配的线程，则将m加1对应的新数组元素重新置为null，将m减1，然后再次for循环抢占其他为null的数组元素。极端并发下m会一直增加直到达到最大值FULL为止，达到FULL后只能通过for循环不断尝试与其他线程匹配或者抢占为null的数组元素，然后随着并发减少，m会一直减少到0。通过这种动态调整m的方式可以避免过多的线程基于CAS修改同一个元素导致CAS失败，提高匹配的效率，这种思想跟LongAdder的实现是一致的。
  • 只有当m等于0的时候才会通过Unsafe park方法让线程休眠，如果不等于0，即此时存在多个并行的等待匹配的线程，则主要通过自旋的方式等待其他线程到来，这是因为交换动作本身是很快的很短暂的，通过自旋等待就可以让多个等待的线程快速的完成匹配;只有当前只剩下一个线程的时候，此时m肯定等于0，短期内没有匹配的线程，才会考虑通过park方法阻塞。
 
 
 
 
    
  
  
  
  
  1. // 抢占slot失败后进入此方法，arena不为空    
    2. 
  
  
  
  
  2. private final Object arenaExchange(Object item, boolean timed, long ns) {
    3. 
  
  
  
  
  3.     Node[] a = arena;
    4. 
  
  
  
  
  4.     Node p = participant.get();
    5. 
  
  
  
  
  5.     // index初始为0
    6. 
  
  
  
  
  6.     for (int i = p.index;;) {                      // access slot at i
    7. 
  
  
  
  
  7.         int b, m, c; long j;                       // j is raw array offset
    8. 
  
  
  
  
  8.         // 在创建arena时，将本来的数组容量 << ASHIFT,为了避免数组元素落到了同一个高速缓存行
    9. 
  
  
  
  
  9.         // 这里获取真实的数组元素索引时也需要 << ASHIFR
    10. 
  
  
  
  
  10.         Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
    11. 
  
  
  
  
  11.         // 如果q不为null，则将对应的数组元素置为null，表示当前线程和该元素对应的线程匹配l
    12. 
  
  
  
  
  12.         if (q != null && U.compareAndSwapObject(a, j, q, null)) {
    13. 
  
  
  
  
  13.             Object v = q.item;                     // release
    14. 
  
  
  
  
  14.             q.match = item; // 保存item，交互成功
    15. 
  
  
  
  
  15.             Thread w = q.parked;
    16. 
  
  
  
  
  16.             if (w != null) // 唤醒等待的线程
    17. 
  
  
  
  
  17.                 U.unpark(w);
    18. 
  
  
  
  
  18.             return v;
    19. 
  
  
  
  
  19.         }
    20. 
  
  
  
  
  20.         // q为null 或者q不为null，cas抢占q失败了
    21. 
  
  
  
  
  21.         // bound初始化时时SEQ，SEQ & MMASK 就是0，即m的初始值就0，m为0时，i肯定为0
    22. 
  
  
  
  
  22.         else if (i <= (m = (b = bound) & MMASK) && q == null) {
    23. 
  
  
  
  
  23.             p.item = item;                         // offer
    24. 
  
  
  
  
  24.             if (U.compareAndSwapObject(a, j, null, p)) {
    25. 
  
  
  
  
  25.                 long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;
    26. 
  
  
  
  
  26.                 Thread t = Thread.currentThread(); // wait
    27. 
  
  
  
  
  27.                 for (int h = p.hash, spins = SPINS;;) {
    28. 
  
  
  
  
  28.                     Object v = p.match;
    29. 
  
  
  
  
  29.                     if (v != null) {
    30. 
  
  
  
  
  30.                         U.putOrderedObject(p, MATCH, null);
    31. 
  
  
  
  
  31.                         p.item = null;             // clear for next use
    32. 
  
  
  
  
  32.                         p.hash = h;
    33. 
  
  
  
  
  33.                         return v;
    34. 
  
  
  
  
  34.                     }
    35. 
  
  
  
  
  35.                     else if (spins > 0) {
    36. 
  
  
  
  
  36.                         h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
    37. 
  
  
  
  
  37.                         if (h == 0)                // initialize hash
    38. 
  
  
  
  
  38.                             h = SPINS | (int)t.getId();
    39. 
  
  
  
  
  39.                         else if (h < 0 &&          // approx 50% true
    40. 
  
  
  
  
  40.                                  (--spins & ((SPINS >>> 1) - 1)) == 0)
    41. 
  
  
  
  
  41.                             Thread.yield();        // two yields per wait
    42. 
  
  
  
  
  42.                     }
    43. 
  
  
  
  
  43.                     else if (U.getObjectVolatile(a, j) != p)
    44. 
  
  
  
  
  44.                         spins = SPINS;       // releaser hasn't set match yet
    45. 
  
  
  
  
  45.                     else if (!t.isInterrupted() && m == 0 &&
    46. 
  
  
  
  
  46.                              (!timed ||
    47. 
  
  
  
  
  47.                               (ns = end - System.nanoTime()) > 0L)) {
    48. 
  
  
  
  
  48.                         U.putObject(t, BLOCKER, this); // emulate LockSupport
    49. 
  
  
  
  
  49.                      &n            
    
                新闻名称：并发编程之Exchanger原理与使用
                
    
                文章转载：http://www.shufengxianlan.com/qtweb/news46/534896.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联

猜你还喜欢下面的内容

响应式网站知识

同城分类信息

定制级高端网站建设　　 　成都网站设计公司　　 　成都app软件开发公司　　 　南充做网站　　 　成都网站建设　　 　成都网站设计　　 　成都温江机房　　 　绵阳网站制作　　 　成都建网站　　 　复盛机电设备　　 　成都网站运营维护　　 　绵阳网站建设公司　　 　百度推广公司　　 　南充网站制作　　 　成都封阳台　　 　小程序开发　　 　犀浦封阳台　　 　网站建设　　 　成都网站推广　　 　成都网站改版