并发编程之-Exchanger-源码分析
JUC 包中除了 CountDownLatch, CyclicBarrier, Semaphore, 还有一个重要的工具,只不过相对而言使用的不多,什么呢? Exchange —— 交换器。用于在两个线程之间交换数据,A 线程将 a 数据交给 B 线程,B 线程将 b 数据交给 a 线程。
具体使用例子参见 并发编程之 线程协作工具类。我们这篇文章就不再讲述如何使用了。
而今天,我们将从源码处分析,Exchange 的实现原理。如果大家看过之前关于 SynchronousQueue 的文章 并发编程之 SynchronousQueue 核心源码分析,就能够看的出来,Exchange 的原理和他很类似。
1. 源码
类 UML:
内部有 2 个内部类: Node , Participant 重写了 ThreadLocal 的 initialValue 方法。
构造方法如下:
1 | public Exchanger() { |
就是创建了一个 ThreadLocal 对象,并设置了初始值,一个 Node 对象。
看看这个 node 对象:
1 | .misc.Contended |
这个 node 对象就是 A ,B 线程实际存储数据的容器。A 线程存在 item 属性上,B 线程存储在 match 线程上,称为匹配。同时,有个线程对象,你应该猜到做什么用处的吧,对,挂起线程的。
和 SynchronousQueue 的区别在于, SynchronousQueue 使用了一个变量来保存数据项,通过 isData 来区别 “存” 操作和 “取” 操作。而 Exchange 使用了 2 个变量,就不用使用 isData 来区分了。
我们再来看看 Exchange 的唯一重要方法 : exchange 方法。
2. exchange 方法源码分析
代码如下:
1 | public V exchange(V x) throws InterruptedException { |
说一下方法的逻辑:
- 如果执行 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 方法源码分析
代码加注释如下:
1 | private final Object slotExchange(Object item, boolean timed, long ns) { |
源码还是有点小长的。简单说说逻辑。
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 这个语义。而多线程并发时,使用数组来控制,每个线程访问数组中不同的槽。
最后,用我们的图收尾吧: