前言
JUC 包中除了 CountDownLatch, CyclicBarrier, Semaphore, 还有一个重要的工具,只不过相对而言使用的不多,什么呢? Exchange —— 交换器。用于在两个线程之间交换数据,A 线程将 a 数据交给 B 线程,B 线程将 b 数据交给 a 线程。
具体使用例子参见 并发编程之 线程协作工具类。我们这篇文章就不再讲述如何使用了。
而今天,我们将从源码处分析,Exchange 的实现原理。如果大家看过之前关于 SynchronousQueue 的文章 并发编程之 SynchronousQueue 核心源码分析,就能够看的出来,Exchange 的原理和他很类似。
1. 源码
类 UML:
内部有 2 个内部类: Node , Participant 重写了 ThreadLocal 的 initialValue 方法。
构造方法如下:
public Exchanger() {
participant = new Participant();
}
static final class Participant extends ThreadLocal<Node> {
public Node initialValue() { return new Node(); }
}
就是创建了一个 ThreadLocal 对象,并设置了初始值,一个 Node 对象。
看看这个 node 对象:
@sun.misc.Contended
static final class Node {
int index; // node 在 arena 数组下标
int bound; // 交换器的最后记录值
int collides; // 记录的 CAS 失败数
int hash; // 伪随机的自旋数
Object item; // 这个线程的数据项
volatile Object match; // 别的线程提供的元素,也就是释放他的线程提供的数据 item
volatile Thread parked; // 当阻塞时,设置此线程,不阻塞的话就不必了(因为会自旋)
}
这个 node 对象就是 A ,B 线程实际存储数据的容器。A 线程存在 item 属性上,B 线程存储在 match 线程上,称为匹配。同时,有个线程对象,你应该猜到做什么用处的吧,对,挂起线程的。
和 SynchronousQueue 的区别在于, SynchronousQueue 使用了一个变量来保存数据项,通过 isData 来区别 “存” 操作和 “取” 操作。而 Exchange 使用了 2 个变量,就不用使用 isData 来区分了。
我们再来看看 Exchange 的唯一重要方法 : exchange 方法。
2. exchange 方法源码分析
代码如下:
public V exchange(V x) throws InterruptedException {
Object v;
Object item = (x == null) ? NULL_ITEM : x; // translate null args
// arena 不是 Null ,返回的却是 null, 说明线程中断了.
// 如果 arena 是 null, 就执行后面的方法.反之,如果不是 null, 执行没有意义.
// 注意,当 slotExchange 有机会被执行,且返回的不是 null, 这个表达式整个就是 false, 下面的表达式就不会执行了.
// 也就是说,当 slot 有效的时候, arena 是没有必要执行的.
if ((arena != null || (v = slotExchange(item, false, 0L)) == null) &&
// 线程中断了,或者返回的是 null. 说明线程中断了
// 如果线程没有中断 ,就执行后面的方法.
((Thread.interrupted() || (v = arenaExchange(item, false, 0L)) == null))){
throw new InterruptedException();
}
return (v == NULL_ITEM) ? null : (V)v;
}
说一下方法的逻辑:
- 如果执行 slotExchange 有结果,就不再执行 arenaExchange.
- 如果 slot 被占用了,就执行 arenaExchange.
返回值是什么呢?返回值就是对方线程的数据项,如果 A 线程先调用,那么 A 线程将数据项存在 item 中,B 线程后调用,则 B 线程将数据存在 match 属性中。
A 返回的是 match 属性,b 返回的是 item 属性。
从该方法中,可以看到,有 2 个重要的方法: slotExchange, arenaExchange。先简单说说这两个方法。
当没有多线程并发操作 Exchange 的时候,使用 slotExchange 就足够了。 slot 是一个 node 对象。
当出现并发了,一个 slot 就不够了,就需要使用一个 node 数组 arena 操作了。
so,我们先看看 slotExchange 方法吧,两个方法的逻辑类似。
3. slotExchange 方法源码分析
代码加注释如下:
private final Object slotExchange(Object item, boolean timed, long ns) {
Node p = participant.get(); // 从 ThreadLocal 中取出 node 对象
Thread t = Thread.currentThread();// 当前线程
if (t.isInterrupted()) // preserve interrupt status so caller can recheck
return null;
for (Node q;;) {// 死循环
// 另一个下线程进入这里, 假设 slot 有值
if ((q = slot) != null) {
// 将 slot 修改为 null
if (U.compareAndSwapObject(this, SLOT, q, null)) {
// 拿到 q 的 item
Object v = q.item;
// 自己的 item 赋值给 match,以让对方线程获取
q.match = item;
// q 线程
Thread w = q.parked;
// slot 的 parked 就是阻塞等待的线程对象.
if (w != null)
U.unpark(w);
// 返回了上一个线程放入的 item
return v;
}
// 如果使用 CAS 修改slot 失败了,说明 slot 被使用了,那就需要创建 arena 数组了
if (NCPU > 1 && bound == 0 &&
U.compareAndSwapInt(this, BOUND, 0, SEQ)) // SEQ == 256; 默认 BOUND == 0
arena = new Node[(FULL + 2) << ASHIFT];// length = (2 + 2) << 7 == 512
}
// 如果 slot 是 null, 但 arena 有值了,说明有线程竞争 slot 了,返回 null, 执行 arenaExchange 逻辑
else if (arena != null)
return null; // caller must reroute to arenaExchange
else {// 第一次循环,给 p node 的 item 赋值
p.item = item;
// 将 slot 赋值赋值为 p
if (U.compareAndSwapObject(this, SLOT, null, p))
// 赋值成功跳出循环
break;
// 如果 CAS 失败,将 p 的值清空,重来
p.item = null;
}
}
// 当走到这里的时候,说明 slot 是 null, 且 arena 不是 null(没有多线程竞争使用 slot),并且成功将 item 放入了 slot 中.
// 这个时候要做的就是阻塞自己,等待对方取出 slot 的数据项,然后重置 slot 的数据和池化对象的数据
// 伪随机数
int h = p.hash;
// 超时时间
long end = timed ? System.nanoTime() + ns : 0L;
// 自旋,默认 1024
int spins = (NCPU > 1) ? SPINS : 1;
Object v;
// 如果这个值不是 null, 说明数据被其他线程拿走了, 并且其他线程将数据赋值给 match 属性,完成了一次交换
while ((v = p.match) == null) {
// 自旋
if (spins > 0) {
// 计算伪随机数
h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
// 如果算出来的是0,就使用线程 ID
if (h == 0)
h = SPINS | (int)t.getId();
// 如果不是0,就将自旋数减一,并且让出 CPU 时间片
else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield();
}
// 如果自旋数不够了,且 slot 还没有得到,就重置自旋数
else if (slot != p)
spins = SPINS;
// 如果 slot == p 了,说明对 slot 赋值成功
// 如果线程没有中断 && 数组不是 null && 没有超时限制
else if (!t.isInterrupted() && arena == null &&
(!timed || (ns = end - System.nanoTime()) > 0L)) {
// 为线程中的 parkBlocker 属性赋值为 Exchange 自己
U.putObject(t, BLOCKER, this);
// node 节点的阻塞线程为当前线程
p.parked = t;
// 如果这个数据还没有被拿走,阻塞自己
if (slot == p)
U.park(false, ns);
// 线程苏醒后,将 p 的阻塞线程属性清空
p.parked = null;
// 将当前线程的 parkBlocker 属性设置成 null
U.putObject(t, BLOCKER, null);
}
// 如果有超时限制,使用 CAS 将 slot 从 p 变成 null,取消这次交换
else if (U.compareAndSwapObject(this, SLOT, p, null)) {
// 如果CAS成功,如果时间到了 && 线程没有中断 : 返回 time_out 对象: 返回 null
v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
// 跳出内层循环
break;
}
}
// 将 p 的 match 属性设置成 null, 表示初始化状态,没有任何匹配 >>> putOrderedObject是putObjectVolatile的内存非立即可见版本.
U.putOrderedObject(p, MATCH, null);
// 重置 item
p.item = null;
// 保留伪随机数,供下次种子数字
p.hash = h;
// 返回
return v;
}
源码还是有点小长的。简单说说逻辑。
Exchange 使用了对象池的技术,将对象保存在 ThreadLocal 中,这个对象(Node)封装了数据项,线程对象等关键数据。
当第一个线程进入的时候,会将数据放到 池化对象中,并赋值给 slot 的 item.并阻塞自己(通常不会立即阻塞,而是使用 yield 自旋一会儿),等待对方取值.
当第二个线程进入的时候,会拿出存储在 slot item 中的值, 然后对 slot 的 match 赋值,并唤醒上次阻塞的线程.
当第一个线程阻塞被唤醒后,说明对方取到值了,就获取 slot 的 match 值, 并重置 slot 的数据和池化对象的数据,并返回自己的数据.
如果超时了,就返回 Time_out 对象.
如果线程中断了,就返回 null.
在该方法中,会返回 2 种结果,一是有效的 item, 二是 null--- 要么是线程竞争使用 slot 了,创建了 arena 数组,要么是线程中断了.
用一幅图来看看具体逻辑,其实还是挺简单的。
当 slot 被别是线程使用了,那么就需要创建一个 arena 的数组了。通过操纵数组里面的元素来实现数据交换。
关于 arenaExchange 方法的源码我就不贴了,有 2 个原因,一个是总体逻辑和 slotExchange 相同,第二个原因则是,其中有一些细节我没有弄懂,就不发出自己写代码注释了,防止误导。但我们已经掌握了 Exchange 的原理。
总结
Exchange 和 SynchronousQueue 类似,都是通过两个线程操作同一个对象实现数据交换,只不过就像我们开始说的,SynchronousQueue 使用的是同一个属性,通过不同的 isData 来区分,多线程并发时,使用了队列进行排队。
Exchange 使用了一个对象里的两个属性,item 和 match,就不需要 isData 属性了,因为在 Exchange 里面,没有 isData 这个语义。而多线程并发时,使用数组来控制,每个线程访问数组中不同的槽。
最后,用我们的图收尾吧: