A synchronization point at which threads can pair and swap elements within pairs. Each thread presents some object on entry to the exchange method, matches with a partner thread, and receives its partner's object on return. An Exchanger may be viewed as a bidirectional form of a SynchronousQueue. Exchangers may be useful in applications such as genetic algorithms and pipeline designs.

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;/*** @Title: ExchangerTest* @Description: Test class for Exchanger* @Company: CSAIR* @Author: lixuanbin* @Creation: 2014年12月14日* @Version:1.0*/
public class ExchangerTest {protected static final Logger log = Logger.getLogger(ExchangerTest.class);private static volatile boolean isDone = false;static class ExchangerProducer implements Runnable {private Exchanger<Integer> exchanger;private static int data = 1;ExchangerProducer(Exchanger<Integer> exchanger) {this.exchanger = exchanger;}@Overridepublic void run() {while (!Thread.interrupted() && !isDone) {for (int i = 1; i <= 3; i++) {try {TimeUnit.SECONDS.sleep(1);data = i;System.out.println("producer before: " + data);data = exchanger.exchange(data);System.out.println("producer after: " + data);} catch (InterruptedException e) {log.error(e, e);}}isDone = true;}}}static class ExchangerConsumer implements Runnable {private Exchanger<Integer> exchanger;private static int data = 0;ExchangerConsumer(Exchanger<Integer> exchanger) {this.exchanger = exchanger;}@Overridepublic void run() {while (!Thread.interrupted() && !isDone) {data = 0;System.out.println("consumer before : " + data);try {TimeUnit.SECONDS.sleep(1);data = exchanger.exchange(data);} catch (InterruptedException e) {log.error(e, e);}System.out.println("consumer after : " + data);}}}/*** @param args*/public static void main(String[] args) {ExecutorService exec = Executors.newCachedThreadPool();Exchanger<Integer> exchanger = new Exchanger<Integer>();ExchangerProducer producer = new ExchangerProducer(exchanger);ExchangerConsumer consumer = new ExchangerConsumer(exchanger);exec.execute(producer);exec.execute(consumer);exec.shutdown();try {exec.awaitTermination(30, TimeUnit.SECONDS);} catch (InterruptedException e) {log.error(e, e);}}
}

consumer before : 0
producer before: 1
consumer after : 1
producer after: 0
consumer before : 0
producer before: 2
producer after: 0
consumer after : 2
consumer before : 0
producer before: 3
producer after: 0
consumer after : 3

class ExchangerProducer implements Runnable {private Exchanger<Set<XXXStatistics>> exchanger;private Set<XXXStatistics> holder;private Date fltDate;private int threshold;ExchangerProducer(Exchanger<Set<XXXStatistics>> exchanger,Set<XXXStatistics> holder, Date fltDate, int threshold) {this.exchanger = exchanger;this.holder = holder;this.fltDate = fltDate;this.threshold = threshold;}@Overridepublic void run() {try {while (!Thread.interrupted() && !isDone) {List<XXXStatistics> temp1 = null;List<XXXStatistics> temp11 = null;for (int i = 0; i < allCities.size(); i++) {try {temp1 = xxxDao.findStatistics1(fltDate, allCities.get(i));temp11 = xxxDao.findStatistics2(fltDate, allCities.get(i),internationalList);if (temp1 != null && !temp1.isEmpty()) {calculationCounter.addAndGet(temp1.size());if (temp11 != null && !temp11.isEmpty()) {// merge two lists into temp1
                                mergeLists(temp1, temp11);temp11.clear();temp11 = null;}// merge temp1 into holder set
                            mergeListToSet(holder, temp1);temp1.clear();temp1 = null;}} catch (Exception e) {log.error(e, e);}// Insert every ${threshold} or the last into database.if (holder.size() >= threshold|| i == (allCities.size() - 1)) {log.info("data collected: \n" + holder);holder = exchanger.exchange(holder);log.info("data submitted");}}// all cities are calculatedisDone = true;}log.info("calculation job done, calculated: "+ calculationCounter.get());} catch (InterruptedException e) {log.error(e, e);}exchanger = null;holder.clear();holder = null;fltDate = null;}
}

class ExchangerConsumer implements Runnable {private Exchanger<Set<XXXStatistics>> exchanger;private Set<XXXStatistics> holder;ExchangerConsumer(Exchanger<Set<XXXStatistics>> exchanger,Set<XXXStatistics> holder) {this.exchanger = exchanger;this.holder = holder;}@Overridepublic void run() {try {List<XXXStatistics> tempList;while (!Thread.interrupted() && !isDone) {holder = exchanger.exchange(holder);log.info("got data: \n" + holder);if (holder != null && !holder.isEmpty()) {try {// insert data into databasetempList = convertSetToList(holder);insertionCounter.addAndGet(xxxDao.batchInsertXXXStatistics(tempList));tempList.clear();tempList = null;} catch (Exception e) {log.error(e, e);}// clear the set
                    holder.clear();} else {log.info("wtf, got an empty list");}log.info("data processed");}log.info("insert job done, inserted: " + insertionCounter.get());} catch (InterruptedException e) {log.error(e, e);}exchanger = null;holder.clear();holder = null;}
}

public boolean calculateStatistics(Date fltDate) {// initializationcalculationCounter.set(0);insertionCounter.set(0);isDone = false;exec = Executors.newCachedThreadPool();Set<XXXStatistics> producerSet = new TreeSet<XXXStatistics>();Set<XXXStatistics> consumerSet = new TreeSet<XXXStatistics>();Exchanger<Set<XXXStatistics>> xc = new Exchanger<Set<XXXStatistics>>();ExchangerProducer producer = new ExchangerProducer(xc, producerSet,fltDate, threshold);ExchangerConsumer consumer = new ExchangerConsumer(xc, consumerSet);// execution
    exec.execute(producer);exec.execute(consumer);exec.shutdown();boolean isJobDone = false;try {// wait for terminationisJobDone = exec.awaitTermination(calculationTimeoutMinutes,TimeUnit.MINUTES);} catch (InterruptedException e) {log.error(e, e);}if (!isJobDone) {// force shutdown
        exec.shutdownNow();log.error("time elapsed for "+ calculationTimeoutMinutes+ " minutes, but still not finished yet, shut it down anyway.");}// clean upexec = null;producerSet.clear();producerSet = null;consumerSet.clear();consumerSet = null;xc = null;producer = null;consumer = null;System.gc();// return the resultif (isJobDone && calculationCounter.get() > 0&& calculationCounter.get() == insertionCounter.get()) {return true;}return false;
}