Java并发工具类(四):线程间交换数据的Exchanger

简介

Exchanger(交换者)是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点,在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据, 如果第一个线程先执行exchange方法,它会一直等待第二个线程也执行exchange,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。

Exchanger的应用场景

1、Exchanger可以用于遗传算法,遗传算法里需要选出两个人作为交配对象,这时候会交换两人的数据,并使用交叉规则得出2个交配结果。
2、Exchanger也可以用于校对工作。比如我们需要将纸制银流通过人工的方式录入成电子银行流水,为了避免错误,采用AB岗两人进行录入,录入到Excel之后,系统需要加载这两个Excel,并对这两个Excel数据进行校对,看看是否录入的一致。代码如下:

 1 public class ExchangerTest {
 2
 3     private static final Exchanger<String> exgr = new Exchanger<String>();
 4
 5     private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
 6
 7     public static void main(String[] args) {
 8
 9         threadPool.execute(new Runnable() {
10             @Override
11             public void run() {
12                 try {
13                     String A = "银行流水A";// A录入银行流水数据
14                     exgr.exchange(A);
15                 } catch (InterruptedException e) {
16                 }
17             }
18         });
19
20         threadPool.execute(new Runnable() {
21             @Override
22             public void run() {
23                 try {
24                     String B = "银行流水B";// B录入银行流水数据
25                     String A = exgr.exchange("B");
26                     System.out.println("A和B数据是否一致:" + A.equals(B) + ",A录入的是:"
27                             + A + ",B录入是:" + B);
28                 } catch (InterruptedException e) {
29                 }
30             }
31         });
32
33         threadPool.shutdown();
34
35     }
36 }

3、这个类在遇到类似生产者和消费者问题时,是非常有用的。来一个非常经典的并发问题:你有相同的数据buffer,一个或多个数据生产者,和一个或多个数据消费者。只是Exchange类只能同步2个线程,所以你只能在你的生产者和消费者问题中只有一个生产者和一个消费者时使用这个类。

在这个指南,你将学习如何使用 Exchanger 类来解决只有一个生产者和一个消费者的生产者和消费者问题。

按照这些步骤来实现下面的例子:

 1 package tool;
 2 import java.util.List;
 3 import java.util.concurrent.Exchanger;
 4
 5 //1. 首先,从实现producer开始吧。创建一个类名为Producer并一定实现 Runnable 接口。
 6 public class Producer implements Runnable {
 7
 8 // 2. 声明 List<String>对象,名为 buffer。这是等等要被相互交换的数据类型。
 9 private List<String> buffer;
10
11 // 3. 声明 Exchanger<List<String>>; 对象,名为exchanger。这个 exchanger 对象是用来同步producer和consumer的。
12 private final Exchanger<List<String>> exchanger;
13
14 // 4. 实现类的构造函数,初始化这2个属性。
15 public Producer(List<String> buffer, Exchanger<List<String>> exchanger) {
16 this.buffer = buffer;
17 this.exchanger = exchanger;
18 }
19
20 // 5. 实现 run() 方法. 在方法内,实现10次交换。
21 @Override
22 public void run() {
23 int cycle = 1;
24 for (int i = 0; i < 10; i++) {            System.out.printf("Producer: Cycle %d\n", cycle);
25
26 // 6. 在每次循环中,加10个字符串到buffer。
27 for (int j = 0; j <10; j++) {
28 String message = "Event " + ((i * 10) + j);
29 System.out.printf("Producer: %s\n", message);
30 buffer.add(message);
31 }
32
33 // 7. 调用 exchange() 方法来与consumer交换数据。此方法可能会抛出InterruptedException 异常, 加上处理代码。
34 try {
35 buffer = exchanger.exchange(buffer);
36 } catch (InterruptedException e) {
37 e.printStackTrace();
38 }
39 System.out.println("Producer: " + buffer.size());
40 cycle++;
41 }
42 }
43 }

 1 //8. 现在, 来实现consumer。创建一个类名为Consumer并一定实现 Runnable 接口。
 2 package tool;
 3 import java.util.List;
 4 import java.util.concurrent.Exchanger;
 5 public class Consumer implements Runnable {
 6
 7 // 9. 声明名为buffer的 List<String>对象。这个对象类型是用来相互交换的。
 8 private List<String> buffer;
 9
10 // 10. 声明一个名为exchanger的 Exchanger<List<String>> 对象。用来同步 producer和consumer。
11 private final Exchanger<List<String>> exchanger;
12
13 // 11. 实现类的构造函数,并初始化2个属性。
14 public Consumer(List<String>buffer, Exchanger<List<String>> exchanger) {
15 this.buffer = buffer;
16 this.exchanger = exchanger;
17 }
18
19 // 12. 实现 run() 方法。在方法内,实现10次交换。
20 @Override
21 public void run() {
22 int cycle = 1;
23 for (int i = 0; i < 10; i++) {
24 System.out.printf("Consumer: Cycle %d\n", cycle);
25
26 // 13. 在每次循环,首先调用exchange()方法来与producer同步。Consumer需要消耗数据。此方法可能会抛出InterruptedException异常, 加上处理代码。
27 try {
28 buffer = exchanger.exchange(buffer);
29 } catch (InterruptedException e) {                e.printStackTrace();
30 }
31
32 // 14. 把producer发来的在buffer里的10字符串写到操控台并从buffer内删除,留空。System.out.println("Consumer: " + buffer.size());
33 for (int j = 0; j <10; j++) {
34 String message = buffer.get(0);
35 System.out.println("Consumer: " + message);
36 buffer.remove(0);
37 }
38 cycle++;
39 }

 1 //15.现在,实现例子的主类通过创建一个类,名为Core并加入 main() 方法。
 2 package tool;
 3 import java.util.ArrayList;
 4 mport java.util.List;
 5 import java.util.concurrent.Exchanger;
 6
 7 public class Core {
 8 public static void main(String[] args) {
 9
10 // 16. 创建2个buffers。分别给producer和consumer使用.
11 List<String> buffer1 = new ArrayList<String>();
12 List<String> buffer2 = new ArrayList<String>();
13
14 // 17. 创建Exchanger对象,用来同步producer和consumer。
15 Exchanger<List<String>> exchanger = new Exchanger<List<String>>();
16
17 // 18. 创建Producer对象和Consumer对象。
18 Producer producer = new Producer(buffer1, exchanger);
19 Consumer consumer = new Consumer(buffer2, exchanger);
20
21 // 19. 创建线程来执行producer和consumer并开始线程。
22 Thread threadProducer = new Thread(producer);
23 Thread threadConsumer = new Thread(consumer); threadProducer.start();
24 threadConsumer.start();
25 }

消费者开始时是空白的buffer,然后调用Exchanger来与生产者同步。因为它需要数据来消耗。生产者也是从空白的buffer开始,然后创建10个字符串,保存到buffer,并使用exchanger与消费者同步。

在这儿,2个线程(生产者和消费者线程)都是在Exchanger里并交换了数据类型,所以当消费者从exchange() 方法返回时,它有10个字符串在buffer内。当生产者从 exchange() 方法返回时,它有空白的buffer来重新写入。这样的操作会重复10遍。

如你执行例子,你会发现生产者和消费者是如何并发的执行任务和在每个步骤它们是如何交换buffers的。与其他同步工具一样会发生这种情况,第一个调用 exchange()方法会进入休眠直到其他线程的达到。

其他方法

如果两个线程有一个没有到达exchange方法,则会一直等待,如果担心有特殊情况发生,避免一直等待,可以使用exchange(V data, long time, TimeUnit unit)设置最大等待时长。

--- V是声明Phaser的参数种类(例子里是 List)。 此线程会休眠直到另一个线程到达并中断它,或者特定的时间过去了。TimeUnit类有多种常量:DAYS, HOURS, MICROSECONDS, MILLISECONDS, MINUTES, NANOSECONDS, 和 SECONDS。

参考:《Java并发编程的艺术》

时间: 2024-12-26 09:42:06

Java并发工具类(四):线程间交换数据的Exchanger的相关文章

Java并发工具类之线程间数据交换工具Exchanger

Exchanger是一个用于线程间协做的工具类,主要用于线程间的数据交换.它提供了一个同步点,在这个同步点,两个线程可以彼此交换数据.两个线程通过exchange方法交换数据,如果一个线程执行exchange方法,它就会等待另一个线程执行exchange方法,当两个线程都到达了同步点,这两个线程就可以交换数据.将本线程产生的数据传送给对方. Exchanger可用于工作的互相校对,比如我们要把线下产生的交易数据通过人工录入的方式添加到系统中,为了避免错误,我们采用AB两人同时录入的方式,当录入完

Java并发工具类(四)线程间交换数据的Exchanger

简介 Exchanger(交换者)是一个用于线程间协作的工具类.Exchanger用于进行线程间的数据交换.它提供一个同步点,在这个同步点两个线程可以交换彼此的数据.这两个线程通过exchange方法交换数据, 如果第一个线程先执行exchange方法,它会一直等待第二个线程也执行exchange,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方. Exchanger的应用场景 1,Exchanger可以用于遗传算法,遗传算法里需要选出两个人作为交配对象,这时

25.大白话说java并发工具类-CountDownLatch,CyclicBarrier,Semaphore,Exchanger

1. 倒计时器CountDownLatch 在多线程协作完成业务功能时,有时候需要等待其他多个线程完成任务之后,主线程才能继续往下执行业务功能,在这种的业务场景下,通常可以使用Thread类的join方法,让主线程等待被join的线程执行完之后,主线程才能继续往下执行.当然,使用线程间消息通信机制也可以完成.其实,java并发工具类中为我们提供了类似"倒计时"这样的工具类,可以十分方便的完成所说的这种业务场景. 为了能够理解CountDownLatch,举一个很通俗的例子,运动员进行跑

Java并发工具类CyclicBarrier

CyclicBarrier同步屏障 java并发工具类中有一个叫做CyclicBarrier的类,与CountDownLatch类似,都可以实现线程间的同步,但是差别是CyclicBarrier是可重置的同步屏障. 想象一个场景,有N个人不同时间走到一扇门,因为门需要N个人合力才能推开,所以人不足N个时,只能阻塞在此,等到N个人都到了之后,可以推开门,继续进行之前的工作.CyclicBarrier就是这扇门. 看看下面的代码,定义了一个线程数为2的,CyclicBarrier,并在主线程和另外一

【多线程与并发】Java并发工具类

主要有两类 ①并发流程控制相关:CountDownLatch.CyclicBarrier.Semaphore ②线程间交换数据相关:Exchanger: CountDownLatch 作用:允许一个或多个线程等待其他线程完成操作 使用步骤: ①定义一个CountDownLatch(称为计数器),并指定等待次数: ②在合适的时机将计数器减1: ③在需要等待所有任务结束的位置,调用await()方法: 根据JDK中的说明文档整理的两个例子: 例子1: public class CountDownLa

Java并发工具类 - CountDownLatch

Java并发工具类 - CountDownLatch 1.简介 CountDownLatch是Java1.5之后引入的Java并发工具类,放在java.util.concurrent包下面 http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/package-summary.html 官方API. CountDownLatch能够使一个或多个线程等待其他线程完成各自的工作后再执行:CountDownLatch是JDK 5+里面

【Java并发工具类】Java并发容器

前言 Java并发包有很大一部分都是关于并发容器的.Java在5.0版本之前线程安全的容器称之为同步容器.同步容器实现线程安全的方式:是将每个公有方法都使用synchronized修饰,保证每次只有一个线程能访问容器的状态.但是这样的串行度太高,将严重降低并发性,当多个线程竞争容器的锁时,吞吐量将严重降低.因此,在Java 5.0版本时提供了性能更高的容器来改进之前的同步容器,我们称其为并发容器. 下面我们先来介绍Java 5.0之前的同步容器,然后再来介绍Java 5.0之后的并发容器. Ja

juc下的并发工具类和线程池

工具类 CountDownLatch 利用它可以实现类似计数器的功能.比如有一个任务A,它要等待其他4个任务执行完毕之后才能执行,此时就可以利用CountDownLatch来实现这种功能了. package com.yjc.juc; import java.util.concurrent.CountDownLatch; public class CountDownLatchDemo { public static void main(String[] args) throws Interrupt

【Java并发工具类】Semaphore

前言 1965年,荷兰计算机科学家Dijkstra提出的信号量机制成为一种高效的进程同步机制.这之后的15年,信号量一直都是并发编程领域的终结者.1980年,管程被提出,成为继信号量之后的在并发编程领域的第二个选择.目前几乎所有的语言都支持信号量机制,Java也不例外.Java中提供了Semaphore并发工具类来支持信号量机制.下面我们就来了解Java实现的信号量机制. 首先介绍信号量模型,然后介绍如何使用,最后使用信号量来实现一个限流器. 信号量模型 信号量模型图(图来自参考[1]): 信号