【源码分析】JUC一Exchanger

末蓝、 2022-06-09 02:52 171阅读 0赞

Exchanger是一种线程间安全交换数据的机制。当线程A调用Exchange对象的exchange()方法后，他会进入阻塞状态，直到线程B也调用了exchange()方法，然后以线程安全的方式交换数据，之后线程A和B继续运行。  
这里先提出两个疑问，带着疑问我们分析一下源码：

1.  可不可以多个线程之间进行数据交换？
2.  两个线程交换的数据是不是必须类型一致呢？

调试进去，你会发现Exchanger的主要逻辑实现在方法doExchange里面。

private Object doExchange(Object item, boolean timed, long nanos) {
            Node me = new Node(item);                 // Create in case occupying
            /** * index是线程ID的hash值映射到0到max之间的一个值, * max的定义如下：可知其默认值为0，也就是说index的初始值为0 * private final AtomicInteger max = new AtomicInteger(); * */
            int index = hashIndex();                  // Index of current slot
            int fails = 0;                            // Number of CAS failures
    
            /** * 这里的无条件的for循环用于自旋 */
            for (;;) {
                Object y;                             // Contents of current slot
                /** * arena是一个Slot数组，初始化语句如下：这里取index=0即第一个元素 * private static final int CAPACITY = 32; * private volatile Slot[] arena = new Slot[CAPACITY]; */
                Slot slot = arena[index];
                /** * 判断slot是否为null，初始状态arena[0]肯定为null，也就是说初始状态，程序会进入该分支 * 调用createSlot方法创建slot，并赋值给arena的index位置的元素。然后会进入下次循环。 * */
                if (slot == null)                     // Lazily initialize slots
                    createSlot(index);                // Continue loop to reread
                /** * 判断slot的值否为null，在非null的情况下（说明该槽里有值，即有其他线程等待交换），然后 * 通过CAS尝试取出该值并把slot（因为slot是AtomicReference类型，值就是其成员value的 * 值）原值赋值为null，CAS操作成功后，当前slot就被释放了，其他线程可以继续使用这个 * slot，当前两个线程通过Node进行交换值。 */
                else if ((y = slot.get()) != null &&  // Try to fulfill
                         slot.compareAndSet(y, null)) {
                    Node you = (Node)y;               // Transfer item
                    /** * 这里通过CAS交换当前slot对应的两个线程（实际上一个slot同时只能有一个线程占据，这 * 里的意思是一个线程刚好遇到当前slot的值不为null，即有线程等待交换）的值，如果成功 * 则返回交换后的值（这里注意返回的是you.item, you是Node类型，Node是 * AtomicReference类型，you.compareAndSet(null, item)这句代码的作用是CAS替换 * 为null的value值，详情可参考AtomicReference的方法compareAndSet），这一点要特 * 别注意（否则后面的代码很难看懂，特别是spinWait）：等待交换的线程把要交换的数据保 * 存在item成员里，而交换线程把要交换的数据保存在value里面；并unpark * 唤醒交换线程；如果失败（被其他线程抢先了），继续下面的判断，这时如果运气好，坑还没 * 被占（当前slot），则在下一个if分支有机会占坑（只是有机会，运气不好也有可能失 * 败）。 */
                    if (you.compareAndSet(null, item)) {
                        LockSupport.unpark(you.waiter);
                        return you.item;
                    }                                 // Else cancelled; continue
                }
                /** * 判断当前slot的值y是否为null（即是否有线程占据改slot），如果为null（即没有线程占 * 据），则通过CAS把自己的值赋给当前slot（即尝试占据当前slot），如果CAS操作成功，判断 * 当前索引是否0，如果为0（即说明当前线程所在的slot是整个slot数组的第一个元素），则阻塞 * 等待（timed标示阻塞是否有超时，nanos是阻塞时间）；如果CAS操作失败（即当前slot被其他 * 线程占据）， */
                else if (y == null &&                 // Try to occupy
                         slot.compareAndSet(null, me)) {
                    if (index == 0)                   // Blocking wait for slot 0
                        return timed ?
                            awaitNanos(me, slot, nanos) :
                            await(me, slot);
                    /** * 所谓的spin wait：就是固定次数循环（详情参考方法spinWait），不同于awaitNanos。 */
                    Object v = spinWait(me, slot);    // Spin wait for non-0
                    if (v != CANCEL)
                        return v;
                    me = new Node(item);              // Throw away cancelled node
                    int m = max.get();
                    /** * 如果spinWait自旋返回CANCEL（即线程没有得到交换的数据被取消，两种可能：一种是被取 * 消，一种是达到自旋次数还未得到交换数据），判断当前最大索引值是否大于当前索引的一半 * 是则把slot数组的最大下标值做减一操作（就好比市场摊位太多，商贩和消费者碰头的概率太 * 低，减少摊位数，增大这个概率）。 */
                    if (m > (index >>>= 1))           // Decrease index
                        max.compareAndSet(m, m - 1);  // Maybe shrink table
                }
                /** * 每个槽上允许2次失败 */
                else if (++fails > 1) {               // Allow 2 fails on 1st slot
                    int m = max.get();
                    /** * CAS失败处理达到3次则增大index，也就是增加CAS的槽的下标（就好比商贩A在市场的当前 * 这个摊位半天没遇到一位顾客，那他肯定觉得这个摊位不发财，要换摊位）。 */
                    if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
                        index = m + 1;                // Grow on 3rd failed slot
                    else if (--index < 0)
                        index = m;                    // Circularly traverse
                }
            }
        }

private static Object spinWait(Node node, Slot slot) {
          /** * private static final int SPINS = (NCPU == 1) ? 0 : 2000; * NCPU 表示CPU的核数， * 所谓的spin wait：就是固定次数循环，每次计数减一对于单核系统来说，spin wait * 是不做的，因为单核做wait时需要占用CPU，其他线程是无法使用CPU，因此这样的等待 * 毫无意义。而多核系统中spin值为2000，也就是会做2000次循环。如果循环完成后依然 * 没得到交换的数据，那么会返回一个CANCEL对象表示请求依旧被取消，并且把Node从 * slot中清除。详情参考spinWait方法。 */
            int spins = SPINS;
            for (;;) {
                /** * 这里需要注意，node.get()返回值是Node（AtomicReference类型）的value，一定要与Node * 的item区分开，当交换线程来到之前，等待交换的线程自旋获取Node的value，直到value不为 * 空（即交换线程到来并交换了数据，在方法doExchange的第二个if分支中交换线程通过 * you.compareAndSet(null, item)，把自己的交换数据item赋值给为null的value）。 */
                Object v = node.get();
                if (v != null)
                    return v;
                else if (spins > 0)
                    --spins;
                else
                    tryCancel(node, slot);
            }
        }