《java.util.concurrent 包源码阅读》18 Exchanger

Exchanger可以看做双向数据传输的SynchronousQueue，即没有生产者和消费者之分，任意两个线程都可以交换数据。

在JDK5中Exchanger被设计成一个容量为1的容器，存放一个等待线程，直到有另外线程到来就会发生数据交换，然后清空容器，等到下一个到来的线程。

从JDK6开始，Exchanger用了类似ConcurrentMap的分段思想，提供了多个slot，增加了并发执行时的吞吐量。

Exchanger不存在公平不公平的模式，因为没有排队的情况发生，只要有两个线程就可以发生数据交换。

直接看核心方法：

private Object doExchange(Object item, boolean timed, long nanos) {
Node me = new Node(item);
// index是线程ID的hash值映射到0到max之间的一个值
// 一般情况下max为0，这样线程交换数据只会使用第一个slot，
// 即index是0，而max不为0情况请看下面的循环
int index = hashIndex();

// CAS操作失败的次数
int fails = 0;

for (;;) {
// 当前slot中存储的对象，也就是Node
Object y;
Slot slot = arena[index];
// 延迟加载，即只有当slot为null时才创建一个slot
// 延迟加载后重新循环一次
if (slot == null)
createSlot(index);
// slot中有数据，也就意味着有线程在等待交换数据
// 这时可以尝试用CAS重置slot(把slot存储的对象设为null)
// 用slot中存储的对象和当前线程进行数据交换
// 如果交换成功就通知原先等待的线程
else if ((y = slot.get()) != null &&
slot.compareAndSet(y, null)) {
Node you = (Node)y;
if (you.compareAndSet(null, item)) {
LockSupport.unpark(you.waiter);
return you.item;
}
// 如果slot存储的对象已经被重置为null，但是数据交换失败了
// 这时就意味着这个等待的线程的交换请求被取消了
// 在分析wait类型的方法代码时会看到如何处理这种情况
}
// 如果slot中没有存储对象，那么首先尝试把当前线程存储到slot中
// 如果存储失败了，就重新循环
else if (y == null &&
slot.compareAndSet(null, me)) {
// index为0意味着仅仅有当前线程在等待交换数据,因此直接等待即可
if (index == 0)
return timed ?
awaitNanos(me, slot, nanos) :
await(me, slot);
// 所谓的spin wait：就是固定次数循环，每次计数减一
// 对于单核系统来说，spin wait是不做的，因为单核
// 做wait时需要占用CPU，其他线程是无法使用CPU，因此这样
// 的等待毫无意义。而多核系统中spin值为2000，也就是会做
// 2000次循环。
// 如果循环完成后依然没有得到交换的数据，那么会返回一个
// CANCEL对象表示请求依旧被取消，并且把Node从slot中清除
Object v = spinWait(me, slot);
if (v != CANCEL)
return v;
// 如果取消了，就新建一个Node取消原先取消的Node用于下次循环
me = new Node(item);
int m = max.get();
// index除2，缩小slot的范围
// 同时如果m过大，减小m
if (m > (index >>>= 1))
max.compareAndSet(m, m - 1);
}
// 允许CAS失败两次，因为两个else if中都有CAS，因此这里
// 允许两个else if的CAS操作都失败过
else if (++fails > 1) {
int m = max.get();
// 失败超过3次，增大m，并且从m处重新索引
if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
index = m + 1;
// 当index小于0，回到m，重新循环
else if (--index < 0)
index = m;
}
}
}

这篇文章关于索引index这块弄得不是很清楚，后续会继续研究，及时更新。