【Java并发编程】22、Exchanger源码解析（JDK1.7…

public class ExchangerDemo {
    public static void main(String[] args) {
        Exchanger<List<Integer>> exchanger = new Exchanger<>();
        new Consumer(exchanger).start();
        //方便调试，让consumer先执行exchange
        try {
            Thread.sleep(1000 * 5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        new Producer(exchanger).start();
    }

    static class Consumer extends Thread {
        List<Integer> list = new ArrayList<>();
        Exchanger<List<Integer>> exchanger = null;

        public Consumer(Exchanger<List<Integer>> exchanger) {
            super();
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            for (int i = 0; i < 1; i++) {
                try {
                    list = exchanger.exchange(list);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.print(list.get(0) + ", ");
                System.out.print(list.get(1) + ", ");
                System.out.print(list.get(2) + ", ");
                System.out.print(list.get(3) + ", ");
                System.out.println(list.get(4) + ", ");
            }
        }
    }

    static class Producer extends Thread {
        List<Integer> list = new ArrayList<>();
        Exchanger<List<Integer>> exchanger = null;

        public Producer(Exchanger<List<Integer>> exchanger) {
            super();
            this.exchanger = exchanger;
        }

        @Override
        public void run() {
            Random rand = new Random();
            for (int i = 0; i < 1; i++) {
                list.clear();
                list.add(rand.nextInt(10000));
                list.add(rand.nextInt(10000));
                list.add(rand.nextInt(10000));
                list.add(rand.nextInt(10000));
                list.add(rand.nextInt(10000));
                try {
                    list = exchanger.exchange(list);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

private static final class Node extends AtomicReference<Object> {  
    /** 创建这个节点的线程提供的用于交换的数据。 */  
    public final Object item;  
    /** 等待唤醒的线程 */  
    public volatile Thread waiter;  
    /** 
     * Creates node with given item and empty hole. 
     * @param item the item 
     */  
    public Node(Object item) {  
        this.item = item;  
    }  
}  
  
/** 
 * 一个Slot就是一对线程交换数据的地方。 
 * 这里对Slot做了缓存行填充，能够避免伪共享问题。 
 * 虽然填充导致浪费了一些空间，但Slot是按需创建，一般没什么问题。 
 */  
private static final class Slot extends AtomicReference<Object> {  
    // Improve likelihood of isolation on <= 64 byte cache lines  
    long q0, q1, q2, q3, q4, q5, q6, q7, q8, q9, qa, qb, qc, qd, qe;  
}  
  
/** 
 * Slot数组，在需要时才进行初始化。 
 * 用volatile修饰，因为这样可以安全的使用双重锁检测方式构建。 
 */  
private volatile Slot[] arena = new Slot[CAPACITY];  
/** 
 * arena(Slot数组)的容量。设置这个值用来避免竞争。 
 */  
private static final int CAPACITY = 32;  
/** 
 * 正在使用的slot下标的最大值。当一个线程经历了多次CAS竞争后， 
 * 这个值会递增；当一个线程自旋等待超时后，这个值会递减。 
 */  
private final AtomicInteger max = new AtomicInteger();  

/** 
 * 等待其他线程到达交换点，然后与其进行数据交换。 
 * 
 * 如果其他线程到来，那么交换数据，返回。 
 * 
 * 如果其他线程未到来，那么当前线程等待，知道如下情况发生： 
 *   1.有其他线程来进行数据交换。 
 *   2.当前线程被中断。 
 */  
public V exchange(V x) throws InterruptedException {  
    if (!Thread.interrupted()) {//检测当前线程是否被中断。  
        //进行数据交换。  
        Object v = doExchange(x == null? NULL_ITEM : x, false, 0);  
        if (v == NULL_ITEM)  
            return null; //检测结果是否为null。  
        if (v != CANCEL) //检测是否被取消。  
            return (V)v;  
        Thread.interrupted(); // 清除中断标记。  
    }  
    throw new InterruptedException();  
}  
/** 
 * 等待其他线程到达交换点，然后与其进行数据交换。 
 *  
 * 如果其他线程到来，那么交换数据，返回。 
 *  
 * 如果其他线程未到来，那么当前线程等待，知道如下情况发生： 
 *   1.有其他线程来进行数据交换。 
 *   2.当前线程被中断。 
 *   3.超时。 
 */  
public V exchange(V x, long timeout, TimeUnit unit)  
    throws InterruptedException, TimeoutException {  
    if (!Thread.interrupted()) {  
        Object v = doExchange(x == null? NULL_ITEM : x,  
                              true, unit.toNanos(timeout));  
        if (v == NULL_ITEM)  
            return null;  
        if (v != CANCEL)  
            return (V)v;  
        if (!Thread.interrupted())  
            throw new TimeoutException();  
    }  
    throw new InterruptedException();  
}  

private Object doExchange(Object item, boolean timed, long nanos) {
        Node me = new Node(item);                 
        int index = hashIndex();                  //根据thread id计算出自己要去的那个交易位置（slot)
        int fails = 0;                            

        for (;;) {
            Object y;                             
            Slot slot = arena[index];
            if (slot == null)                    
                createSlot(index);     //slot = null，创建一个slot，然后会回到for循环，再次开始
            else if ((y = slot.get()) != null &&  //slot里面有人等着(有Node)，则尝试和其交换
                     slot.compareAndSet(y, null)) { //关键点1：slot清空，Node拿出来，俩人在Node里面交互。把Slot让给后面的人，做交互地点
                Node you = (Node)y;               
                if (you.compareAndSet(null, item)) {//把Node里面的东西，换成自己的
                    LockSupport.unpark(you.waiter); //唤醒对方
                    return you.item; //自己把对方的东西拿走
                } //关键点2：如果你运气不好，在Node里面要交换的时候，被另一个线程抢了，回到for循环，重新开始                          
            }
            else if (y == null &&                 //slot里面为空(没有Node)，则自己把位置占住
                     slot.compareAndSet(null, me)) {
                if (index == 0)                   //如果是0这个位置，自己阻塞，等待别人来交换
                    return timed? awaitNanos(me, slot, nanos): await(me, slot);
                Object v = spinWait(me, slot);    //不是0这个位置，自旋等待
                if (v != CANCEL)  //自旋等待的时候，运气好，有人来交换了，返回
                    return v;
                me = new Node(item);     //自旋的时候，没人来交换。走执行下面的，index减半，挪个位置，重新开始for循环    
                int m = max.get();
                if (m > (index >>>= 1))           
                    max.compareAndSet(m, m - 1); 
            }
            else if (++fails > 1) { //失败 case1: slot有人，要交互，但被人家抢了  case2: slot没人，自己要占位置，又被人家抢了    
                int m = max.get();
                if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
                    index = m + 1;   //3次匹配失败，把index扩大，再次开始for循环             
                else if (--index < 0)
                    index = m;                   
            }
        }
    }