j.u.c系列(11)---之并发工具类:Exchanger

时间:2022-10-30 18:01:31

写在前面

  前面三篇博客分别介绍了CyclicBarrier、CountDownLatch、Semaphore,现在介绍并发工具类中的最后一个Exchange。Exchange是最简单的也是最复杂的,简单在于API非常简单,就一个构造方法和两个exchange()方法,最复杂在于它的实现是最复杂的(本文不解释。太复杂)

  在API是这么介绍的:可以在对中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给 exchange 方法,与伙伴线程进行匹配,并且在返回时接收其伙伴的对象。Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序(比如遗传算法和管道设计)中很有用。

  Exchanger,它允许在并发任务之间交换数据。具体来说,Exchanger类允许在两个线程之间定义同步点。当两个线程都到达同步点时,他们交换数据结构,因此第一个线程的数据结构进入到第二个线程中,第二个线程的数据结构进入到第一个线程中。

 

应用示例

public class ExchangerTest {

private static final Exchanger<String> exgr = new Exchanger<String>();

private static ExecutorService threadPool = Executors.newFixedThreadPool(2);

public static void main(String[] args) {

threadPool.execute(
new Runnable() {
@Override
public void run() {
try {
String A
= "银行流水A";// A录入银行流水数据
exgr.exchange(A);
}
catch (InterruptedException e) {
}
}
});

threadPool.execute(
new Runnable() {
@Override
public void run() {
try {
String B
= "银行流水B";// B录入银行流水数据
String A = exgr.exchange("B");
System.out.println(
"A和B数据是否一致:" + A.equals(B) + ",A录入的是:"
+ A + ",B录入是:" + B);
}
catch (InterruptedException e) {
}
}
});

threadPool.shutdown();

}
}

示例2

public class ExchangerTest {
static class Producer implements Runnable{
//生产者、消费者交换的数据结构
private List<String> buffer;
//步生产者和消费者的交换对象
private Exchanger<List<String>> exchanger;
Producer(List
<String> buffer,Exchanger<List<String>> exchanger){
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for(int i = 1 ; i < 5 ; i++){
System.out.println(
"生产者第" + i + "次提供");
for(int j = 1 ; j <= 3 ; j++){
System.out.println(
"生产者装入" + i + "--" + j);
buffer.add(
"buffer:" + i + "--" + j);
}
System.out.println(
"生产者装满,等待与消费者交换...");
try {
exchanger.exchange(buffer);
}
catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

static class Consumer implements Runnable {
private List<String> buffer;
private final Exchanger<List<String>> exchanger;
public Consumer(List<String> buffer, Exchanger<List<String>> exchanger) {
this.buffer = buffer;
this.exchanger = exchanger;
}
@Override
public void run() {
for (int i = 1; i < 5; i++) {
//调用exchange()与消费者进行数据交换
try {
buffer
= exchanger.exchange(buffer);
}
catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(
"消费者第" + i + "次提取");
for (int j = 1; j <= 3 ; j++) {
System.out.println(
"消费者 : " + buffer.get(0));
buffer.remove(
0);
}
}
}
}
public static void main(String[] args){
List
<String> buffer1 = new ArrayList<String>();
List
<String> buffer2 = new ArrayList<String>();
Exchanger
<List<String>> exchanger = new Exchanger<List<String>>();

Thread producerThread
= new Thread(new Producer(buffer1,exchanger));
Thread consumerThread
= new Thread(new Consumer(buffer2,exchanger));
producerThread.start();
consumerThread.start();
}
}
生产者第1次提供
生产者装入1
--1
生产者装入1
--2
生产者装入1
--3
生产者装满,等待与消费者交换...
生产者第2次提供
消费者第1次提取
生产者装入2
--1
消费者 : buffer:
1--1
生产者装入2
--2
生产者装入2
--3
生产者装满,等待与消费者交换...
消费者 : buffer:
1--2
消费者 : buffer:
1--3
消费者第2次提取
消费者 : buffer:
2--1
消费者 : buffer:
2--2
消费者 : buffer:
2--3
生产者第3次提供
生产者装入3
--1
生产者装入3
--2
生产者装入3
--3
生产者装满,等待与消费者交换...
生产者第4次提供
生产者装入4
--1
生产者装入4
--2
生产者装入4
--3
生产者装满,等待与消费者交换...
消费者第3次提取
消费者 : buffer:
3--1
消费者 : buffer:
3--2
消费者 : buffer:
3--3
消费者第4次提取
消费者 : buffer:
4--1
消费者 : buffer:
4--2
消费者 : buffer:
4--3