lchl0388 发表于 2017-7-2 13:17:47

thread_Exchanger数据交换

  Exchanger 是一个同步辅助类,用于两个并发线程之间在一个同步点进行数据交换。
允许两个线程在某一个点进行数据交换。
可以视作双向的同步队列;
可应用于基因算法、流水线设计等场景
  Exchanger提供了 一个同步点 , 在这个同步点,两个线程可以交换数据 。每个线程通过exchange()方法的入口提供数据给另外的线程,并接收其它线程提供的数据,并返回。



public class Exchanger1Test {
// 场景描述:一对一的 生产者和消费者,生产者每次生产5个商品,然后消费者把空的商品容器和生产者交换。
// 生产者线程一定要先生产数据,再交换数据,消费者线程一定要先交换数据,再消费数据,否则会出现少消费数据的现象
// 允许原子性的交换两个(多个)对象,但同时只有一对才会成功
// exchange方法真的帮一对线程交换了数据;
// exchange方法真的会阻塞调用方线程直至另一方线程参与交易。
public static void main(String[] args) {
// Exchanger可以在两个线程之间交换数据,只能是2个线程,他不支持更多的线程之间互换数据。
// 两个线程必须使用同一个Exchanger对象,且只能是两个线程间的数据交换
// 当线程A调用Exchange对象的exchange()方法后,他会陷入阻塞状态,直到线程B也调用了exchange()方法,然后以线程安全的方式交换数据,之后线程A和B继续运行
Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new Producer(exchanger));
exec.execute(new Consumer(exchanger));
exec.shutdown();
try {
exec.awaitTermination(30, TimeUnit.SECONDS);
} catch (InterruptedException e) {
// log.error(e, e);
      }   
}
}
// 生产者
class Producer implements Runnable {
private ArrayList<String> goods = new ArrayList<String>(); // 商品容器
private Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
//控制交易双方线程的退出
private static AtomicBoolean isDone = new AtomicBoolean(true);
public Producer(Exchanger<ArrayList<String>> exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
while (!Thread.interrupted() &&isDone.get()) {//
            
for (int i = 0; i < 3; i++) { // 生产3次
System.out.println("------------------------生产者生产第 " + i + "次");
for (int j = 0; j < 3; j++) { // 每次生产3个商品
String e = (long) (Math.random() * 1000) + "";
goods.add(e);
System.out.println("生产了商品:" + e);
}
try {
// 生产者线程一定要先生产数据,再交换数据,消费者线程一定要先交换数据,再消费数据
// exchanger.exchange(v)的时候,当前线程会被阻塞,直到另一个线程执行该方法,同时完成数据的交换
goods = exchanger.exchange(goods); // 交换数据
System.out.println("生产者:数据交换完毕:获得交换的商品容器大小:" + goods.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
// 消费者
class Consumer implements Runnable {
private ArrayList<String> goods = new ArrayList<String>(); // 商品容器
private Exchanger<ArrayList<String>> exchanger = new Exchanger<ArrayList<String>>();
private static AtomicBoolean isDone = new AtomicBoolean(true);
public Consumer(Exchanger<ArrayList<String>> exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
while (!Thread.interrupted() && isDone.get()) {//&& !isDone
for (int i = 0; i < 3; i++) { // 消费3次
try {
// 生产者线程一定要先生产数据,再交换数据,消费者线程一定要先交换数据,再消费数据
// exchanger.exchange(v)的时候,当前线程会被阻塞,直到另一个线程执行该方法,同时完成数据的交换
goods = exchanger.exchange(goods); // 交换数据
System.out.println("消费者:数据交换完毕:获得交换的商品容器大小:" + goods.size());
// 消费商品
Iterator<String> it = goods.iterator();
if (goods.size() > 0) {
System.out.println("*********************消费者消费第 " + i + "次");
while (it.hasNext()) {
String next = it.next();
System.out.println("消费了商品:" + next);
it.remove(); // 移除消费了的商品
                        }
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
页: [1]
查看完整版本: thread_Exchanger数据交换