高并发编程之Exchanger工具类讲解

一、Exchanger介绍
①、Exchanger，并发工具类，用于线程间的数据交换。
②、两个线程，两个缓冲区，一个线程往一个缓冲区里面填数据，另一个线程从另一个缓冲区里面取数据。当填数据的线程将缓冲区填满时，或者取数据的线程将缓冲区里的数据取空时，就主动向对方发起交换缓冲区的动作，而交换的时机是，一个缓冲区满，另一个缓冲区空。

注意：使用Exchanger来对线程进行数据操作时，线程必须是成对的（线程数量为双数）。

二、介绍Exchanger两个重要方法
①、exchange(V x)：等待另一个线程到达此交换点（除非当前线程被中断），然后将给定的对象传送给该线程，并接收该线程的对象。

package chapter3.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* @author czd
*/
public class ExchangerTest1 {
public static void main(String[] args) {
final Exchanger<String> exchanger = new Exchanger<>();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
String result = exchanger.exchange(" 我是来自Thread-A发送的数据！");
System.out.println(Thread.currentThread().getName() + " Result: " + result);
}catch (InterruptedException e){
e.printStackTrace();
}

}
},"Thread-A").start();

new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
TimeUnit.SECONDS.sleep(5);
String result = exchanger.exchange(" 我是来自Thread-B发送的数据！");
System.out.println(Thread.currentThread().getName() + " Result: " + result);
}catch (Exception e){
e.printStackTrace();
}

}
},"Thread-B").start();
}
}

输出结果

②、exchange(V x, long timeout, TimeUnit unit)：等待另一个线程到达此交换点（除非当前线程被中断，或者超出了指定的等待时间），然后将给定的对象传送给该线程，同时接收该线程的对象。

package chapter3.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* @author czd
*/
public class ExchangerTest1 {
public static void main(String[] args) {
final Exchanger<String> exchanger = new Exchanger<>();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
String result = exchanger.exchange(" 我是来自Thread-A发送的数据！",3,TimeUnit.SECONDS);
System.out.println(Thread.currentThread().getName() + " Result: " + result);
}catch (InterruptedException e){
e.printStackTrace();
} catch (TimeoutException e){
System.out.println("超时了！");
}

}
},"Thread-A").start();

new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
TimeUnit.SECONDS.sleep(5);
String result = exchanger.exchange(" 我是来自Thread-B发送的数据！");
System.out.println(Thread.currentThread().getName() + " Result: " + result);
}catch (Exception e){
e.printStackTrace();
}

}
},"Thread-B").start();
}
}

输出结果

三、提出疑问–A：当使用Exchanger对两个线程进行数据交换时，线程A发送的东西，与线程B接收到的东西是否是一样的？（一样是指地址是否正确，而不仅仅是值一样）

下面是一个验证地址是否一样的代码案例

package chapter3.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
* @author czd
*/
public class ExchangerTest2 {
public static void main(String[] args) {
final Exchanger<Object> exchanger = new Exchanger<>();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
Object object = new Object();
Object obj = exchanger.exchange(object);
System.out.println(Thread.currentThread().getName() + " Thread-A send Object: " + object);
System.out.println(Thread.currentThread().getName() + " Thread-A get Object: " + obj);
}catch (InterruptedException e){
e.printStackTrace();
}

}
},"Thread-A").start();

new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
try {
TimeUnit.SECONDS.sleep(5);
Object object = new Object();
Object obj = exchanger.exchange(object);
System.out.println(Thread.currentThread().getName() + " Thread-B send Object: " + object);
System.out.println(Thread.currentThread().getName() + " Thread-B get Object: " + obj);
}catch (Exception e){
e.printStackTrace();
}

}
},"Thread-B").start();
}
}

输出结果

由图可以看出，线程A发送的东西与线程B接收到的东西是一样的，即连地址都是一样的

四、提出疑问–B：当使用Exchanger对两个线程进行数据交换时，是不是只能交换一次？能否多次的对两个线程进行数据的交换？

下面是一个验证是否能多次进行数据交换的代码案例

package chapter3.exchanger;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
* @author czd
*/
public class ExchangerTest3 {
public static void main(final String[] args) {
final Exchanger<Integer> exchanger = new Exchanger<>();
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "开始启动....");
while (true){
try {
AtomicInteger atomicInteger = new AtomicInteger(1);
atomicInteger.set(exchanger.exchange(atomicInteger.get()));
System.out.println("Thread-A Value:" + atomicInteger.get());
}catch (Exception e){
e.printStackTrace();
}
}

}
},"Thread-A").start();

new Thread(new Runnable() {
@Override
public void run() {

20000
System.out.println(Thread.currentThread().getName() + "开始启动....");
while (true){
try {
AtomicInteger atomicInteger = new AtomicInteger(2);
atomicInteger.set(exchanger.exchange(atomicInteger.get()));
System.out.println("Thread-B Value:" + atomicInteger.get());
TimeUnit.SECONDS.sleep(2);
}catch (Exception e){
e.printStackTrace();
}
}

}
},"Thread-B").start();
}
}

输出结果

由图可以看出，使用Exchanger可以对两个线程进行多次的数据交换