CountDownLatch 同步倒數(shù)計(jì)數(shù)器

CountDownLatch是一個(gè)同步倒數(shù)計(jì)數(shù)器。CountDownLatch允許一個(gè)或多個(gè)線(xiàn)程等待其他線(xiàn)程完成操作。
CountDownLatch對(duì)象內(nèi)部存有一個(gè)整數(shù)作為計(jì)數(shù)器。調(diào)用countDown()方法就將計(jì)數(shù)器減1，當(dāng)計(jì)數(shù)到達(dá)0時(shí)，則所有等待者會(huì)停止等待。計(jì)數(shù)器的操作是原子性的。
CountDownLatch類(lèi)的常用API
構(gòu)造方法
*CountDownLatch(int count) * 構(gòu)造方法參數(shù)指定了計(jì)數(shù)的次數(shù)。
方法
void await() 使當(dāng)前線(xiàn)程在鎖存器倒計(jì)數(shù)至0之前一直等待，除非線(xiàn)程被中斷。
boolean await(long timeout, TimeUnit unit) 使當(dāng)前線(xiàn)程在鎖存器倒計(jì)數(shù)至0之前一直等待，除非線(xiàn)程被中斷或超出了指定的等待時(shí)間。
void countDown() 計(jì)數(shù)減1。當(dāng)計(jì)數(shù)為0，則釋放所有等待的線(xiàn)程。
long getCount() 返回當(dāng)前計(jì)數(shù)。
String toString() 返回標(biāo)識(shí)此鎖存器及其狀態(tài)的字符串。
用給定的計(jì)數(shù)初始化 CountDownLatch實(shí)例。每調(diào)用一次countDown()方法，計(jì)數(shù)器減1。計(jì)數(shù)器大于0 時(shí)，await()方法會(huì)阻塞其他線(xiàn)程繼續(xù)執(zhí)行。 利用該特性，可以讓主線(xiàn)程等待子線(xiàn)程的結(jié)束。

需要注意的是，一旦CountDownLatch的計(jì)數(shù)到0，則無(wú)法再將該計(jì)數(shù)無(wú)法被重置。

一種典型的場(chǎng)景就是火箭發(fā)射。在火箭發(fā)射前，為了保證萬(wàn)無(wú)一失，往往還要進(jìn)行各項(xiàng)設(shè)備、儀器的檢查。只有等所有檢查完畢后，引擎才能點(diǎn)火。這種場(chǎng)景就非常適合使用CountDownLatch。它可以使得點(diǎn)火線(xiàn)程，等待所有檢查線(xiàn)程全部完工后，再執(zhí)行。
例：有三個(gè)工人在為老板干活。老板有一個(gè)習(xí)慣，當(dāng)三個(gè)工人把一天的活都干完了的時(shí)候，他就來(lái)檢查所有工人所干的活。如下代碼設(shè)計(jì)兩個(gè)類(lèi)，Worker代表工人，Boss代表老板。

import java.util.Random; import java.util.concurrent.CountDownLatch; 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; 
import java.util.concurrent.TimeUnit; 
public class CountDownLatchDemo { 
public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        CountDownLatch latch = new CountDownLatch(3);    // 同步倒數(shù)計(jì)數(shù)器。
 Worker w1 = new Worker(latch, "張三");
        Worker w2 = new Worker(latch, "李四");
        Worker w3 = new Worker(latch, "王五");
        Boss boss = new Boss(latch);

        executor.execute(w3); // 工人工作。
 executor.execute(w2);
        executor.execute(w1);
        executor.execute(boss); // 老板工作。
 executor.shutdown();
    }
}
 class Worker implements Runnable { 
private CountDownLatch downLatch; 
private String name; 
public Worker(CountDownLatch downLatch, String name) { 
this.downLatch = downLatch; 
this.name = name;
 } 
public void run() { 
        this.doWork();    // 工人工作。
        try {
            TimeUnit.SECONDS.sleep(new Random().nextInt(10));  // 工作時(shí)長(zhǎng)。
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this.name + "活干完了！"); 
        this.downLatch.countDown();  // 計(jì)數(shù)減1。
 } private void doWork() {
        System.out.println(this.name + "正在干活!");
    }
} 
class Boss implements Runnable { 
private CountDownLatch downLatch; 
public Boss(CountDownLatch downLatch) { 
this.downLatch = downLatch;
    } public void run() {
        System.out.println("老板正在等所有的工人干完活......"); try { this.downLatch.await();    // 當(dāng)計(jì)數(shù)不為0時(shí)，線(xiàn)程永遠(yuǎn)阻塞。為0則繼續(xù)執(zhí)行。
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("工人活都干完了，老板開(kāi)始檢查了！");
    }
}

CountDownLatch類(lèi)與join方法

CountDownLatch實(shí)例本質(zhì)與Thread的join方法相同。但join方法僅可以支持當(dāng)前線(xiàn)程等待一個(gè)線(xiàn)程的結(jié)束，若需要等待多個(gè)線(xiàn)程，則需要逐個(gè)線(xiàn)程的調(diào)用join方法，非常麻煩。CountDwonLatch可以很方便的實(shí)現(xiàn)一個(gè)線(xiàn)程等待多個(gè)線(xiàn)程。

CyclicBarrier 循環(huán)屏障

CyclicBarrier用于讓一組線(xiàn)程運(yùn)行并互相等待，直到共同到達(dá)一個(gè)公共屏障點(diǎn) (common barrier point，又被稱(chēng)為同步點(diǎn))，被屏障攔截的所有線(xiàn)程就會(huì)繼續(xù)執(zhí)行。

CyclicBarrier與CountDownLatch的功能非常類(lèi)似。但一個(gè)CyclicBarrier實(shí)例在釋放等待線(xiàn)程后可以繼續(xù)使用。讓下一批線(xiàn)程在屏障點(diǎn)等待。但CountDownLatch實(shí)例只能被使用一次。所以CyclicBarrier被稱(chēng)為*循環(huán) *的 barrier。

典型的比如公司的人員利用集體郊游，先各自從家出發(fā)到公司集合，再同時(shí)出發(fā)游玩，在指定地點(diǎn)集合。CyclicBarrier表示大家彼此在某處等待，集合好后才開(kāi)始出發(fā)，分散活動(dòng)后又在指定地點(diǎn)集合碰面。

CyclicBarrier類(lèi)API

構(gòu)造器

CyclicBarrier(int parties) 創(chuàng)建CyclicBarrier對(duì)象，parties 表示屏障攔截的線(xiàn)程數(shù)量。

CyclicBarrier(int parties, Runnable barrierAction) 創(chuàng)建 CyclicBarrier對(duì)象，該構(gòu)造方法提供了一個(gè)Runnable 參數(shù)，在一組線(xiàn)程中的最后一個(gè)線(xiàn)程到達(dá)之后，執(zhí)行Runnable中的程序，再之后釋放正在等待的線(xiàn)程。Runnable在屏障點(diǎn)上只運(yùn)行一次。

方法

int await() 通知CyclicBarrier實(shí)例，當(dāng)前線(xiàn)程已經(jīng)到達(dá)屏障點(diǎn)，然后當(dāng)前線(xiàn)程將被阻塞。

int await(long timeout, TimeUnit unit) 指定當(dāng)前線(xiàn)程被阻塞的時(shí)間。

int getNumberWaiting() 返回當(dāng)前在屏障處等待的線(xiàn)程數(shù)。

int getParties() 返回CyclicBarrier的需要攔截的線(xiàn)程數(shù)。

boolean isBroken() 查詢(xún)此屏障是否處于損壞狀態(tài)。

void reset() 將屏障重置為其初始狀態(tài)。

例1：各省數(shù)據(jù)獨(dú)立，分庫(kù)存?zhèn)?。為了提高?jì)算性能，統(tǒng)計(jì)時(shí)采用每個(gè)省開(kāi)一個(gè)線(xiàn)程先計(jì)算單省結(jié)果，最后匯總。

import java.util.concurrent.BrokenBarrierException; 
import java.util.concurrent.CyclicBarrier; 
public class Total { 
public static void main(String[] args) {
        TotalService totalService = new TotalServiceImpl();
        CyclicBarrier barrier = new CyclicBarrier(5, new TotalTask(totalService)); // 實(shí)際系統(tǒng)是查出所有省編碼code的列表，然后循環(huán)，每個(gè)code生成一個(gè)線(xiàn)程。
        new BillTask(new BillServiceImpl(), barrier, "北京").start(); 
        new BillTask(new BillServiceImpl(), barrier, "上海").start(); 
        new BillTask(new BillServiceImpl(), barrier, "廣西").start(); 
        new BillTask(new BillServiceImpl(), barrier, "四川").start(); 
        new BillTask(new BillServiceImpl(), barrier, "黑龍江").start();
    }

} /** * 主任務(wù)：匯總?cè)蝿?wù) */
class TotalTask implements Runnable { 
    private TotalService totalService;

    TotalTask(TotalService totalService) { 
    this.totalService = totalService;
    } public void run() { 
    // 讀取內(nèi)存中各省的數(shù)據(jù)匯總，過(guò)程略。
     totalService.count();
        System.out.println("開(kāi)始全國(guó)匯總");
    }

} /** * 子任務(wù)：計(jì)費(fèi)任務(wù) */
class BillTask extends Thread { 
    private BillService billService;     // 計(jì)費(fèi)服務(wù)
    private CyclicBarrier barrier; 
    private String code;    // 代碼，按省代碼分類(lèi)，各省數(shù)據(jù)庫(kù)獨(dú)立。
BillTask(BillService billService, CyclicBarrier barrier, String code) { 
    this.billService = billService; 
    this.barrier = barrier; 
    this.code = code;
    } 
public void run() {
        System.out.println("開(kāi)始計(jì)算--" + code + "省--數(shù)據(jù)！");
        billService.bill(code); // 把bill方法結(jié)果存入內(nèi)存，如ConcurrentHashMap,vector等,代碼略
        System.out.println(code + "省已經(jīng)計(jì)算完成,并通知匯總Service！"); 
        try { // 通知barrier已經(jīng)完成
        barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
} 
interface BillService { 
      public void bill(String code);
} 
interface TotalService { 
      public void count();
} 
class BillServiceImpl implements BillService{

    @Override 
    public void bill(String code) {}
} 
class TotalServiceImpl implements TotalService{

    @Override 
    public void count(){}
}

例2：賽跑時(shí)，等待所有人都準(zhǔn)備好時(shí)，才起跑。

public class CyclicBarrierTest { public static void main(String[] args) throws IOException, InterruptedException {
        CyclicBarrier barrier = new CyclicBarrier(3);
        ExecutorService executor = Executors.newFixedThreadPool(3);

        executor.submit(new Thread(new Runner(barrier, "1號(hào)選手")));
        executor.submit(new Thread(new Runner(barrier, "2號(hào)選手")));
        executor.submit(new Thread(new Runner(barrier, "3號(hào)選手")));

        executor.shutdown();
    }
} class Runner implements Runnable { // 一個(gè)同步輔助類(lèi)，它允許一組線(xiàn)程互相等待，直到到達(dá)某個(gè)公共屏障點(diǎn) (common barrier point)
    private CyclicBarrier barrier; private String name; 
    public Runner(CyclicBarrier barrier, String name) { 
    super(); 
    this.barrier = barrier; 
    this.name = name;
    }

    @Override 
    public void run() { 
    try {
            Thread.sleep(1000 * (new Random()).nextInt(8));
            System.out.println(name + " 準(zhǔn)備好了..."); // barrier的await方法，在所有參與者都已經(jīng)在此           barrier 上調(diào)用 await 方法之前，將一直等待。
       barrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println(name + " 起跑！");
    }

}

例3：JDK6中的示例用法：下面是一個(gè)在并行分解設(shè)計(jì)中使用 barrier 的例子。給出示意代碼結(jié)構(gòu)，不可運(yùn)行。

class Solver { 
final int N; final float[][] data; 
final CyclicBarrier barrier; 
class Worker implements Runnable { 
int myRow; 
public Worker(int row) {
            myRow = row;
        } public void run() { while (!done()) {
                processRow(myRow); try {
                    barrier.await();
                } catch (InterruptedException ex) { return;
                } catch (BrokenBarrierException ex) { return;
                }
            }
        }

    } public Solver(float[][] matrix) {
        data = matrix;
        N = matrix.length;

        barrier = new CyclicBarrier(N, new Runnable() { public void run() {
                mergeRows(...);
            }
        }); for (int i = 0; i < N; ++i){ new Thread(new Worker(i)).start();
        }
        waitUntilDone();
    }
}

在這個(gè)例子中，每個(gè) worker 線(xiàn)程處理矩陣的一行，在處理完所有的行之前，該線(xiàn)程將一直在屏障處等待。處理完所有的行之后，將執(zhí)行所提供的 Runnable 屏障操作，并合并這些行。如果合并者確定已經(jīng)找到了一個(gè)解決方案，那么 done() 將返回 true，所有的 worker 線(xiàn)程都將終止。

如果屏障操作在執(zhí)行時(shí)不依賴(lài)于正掛起的線(xiàn)程，則線(xiàn)程組中的任何線(xiàn)程在獲得釋放時(shí)都能執(zhí)行該操作。為方便此操作，每次調(diào)用 await() 都將返回能到達(dá)屏障處的線(xiàn)程的索引。然后，可以選擇哪個(gè)線(xiàn)程應(yīng)該執(zhí)行屏障操作，例如：

對(duì)于失敗的同步嘗試，CyclicBarrier 使用了一種要么全部要么全不 (all-or-none) 的破壞模式：如果因?yàn)橹袛?、失敗或者超時(shí)等原因，導(dǎo)致線(xiàn)程過(guò)早地離開(kāi)了屏障點(diǎn)，那么在該屏障點(diǎn)等待的其他所有線(xiàn)程也將通過(guò) BrokenBarrierException（如果它們幾乎同時(shí)被中斷，則用 InterruptedException）以反常的方式離開(kāi)。

Semaphore信號(hào)量

Semaphore用于控制并發(fā)線(xiàn)程數(shù)。Semaphore實(shí)例可以控制當(dāng)前訪問(wèn)自身的線(xiàn)程個(gè)數(shù)。使用Semaphore可以控制同時(shí)訪問(wèn)資源的線(xiàn)程個(gè)數(shù)。例如，實(shí)現(xiàn)一個(gè)文件允許的并發(fā)訪問(wèn)數(shù)。

Semaphore維護(hù)了一個(gè)許可集?！霸S可”即線(xiàn)程進(jìn)入臨界區(qū)的許可。一個(gè)臨界區(qū)可以有多個(gè)許可。獲取許可的線(xiàn)程即可進(jìn)入。通過(guò) acquire() 獲取一個(gè)許可，如果線(xiàn)程沒(méi)有獲取到就等待，而 release() 表示釋放一個(gè)許可。可以把Semaphore看成是一種共享鎖。Semaphore允許同一時(shí)間多個(gè)線(xiàn)程同時(shí)訪問(wèn)臨界區(qū)。

生活的理解：Semaphore實(shí)現(xiàn)的功能就類(lèi)似廁所有5個(gè)坑，假如有十個(gè)人要上廁所，那么同時(shí)能有多少個(gè)人去上廁所呢？同時(shí)只能有5個(gè)人能夠占用，當(dāng)5個(gè)人中的任何一個(gè)人讓開(kāi)后，其中在等待的另外5個(gè)人中又有一個(gè)可以占用了。另外等待的5個(gè)人中可以是隨機(jī)獲得優(yōu)先機(jī)會(huì)，也可以是按照先來(lái)后到的順序獲得機(jī)會(huì)，這取決于構(gòu)造Semaphore對(duì)象時(shí)傳入的參數(shù)選項(xiàng)。

Semaphore對(duì)象也可以實(shí)現(xiàn)互斥鎖的功能，并且可以是由一個(gè)線(xiàn)程獲得了"鎖"，再由另一個(gè)線(xiàn)程釋放"鎖"，這可應(yīng)用于死鎖恢復(fù)的一些場(chǎng)合。

在一些企業(yè)系統(tǒng)中，開(kāi)發(fā)人員經(jīng)常需要限制未處理的特定資源請(qǐng)求（線(xiàn)程/操作）數(shù)量，事實(shí)上，限制有時(shí)候能夠提高系統(tǒng)的吞吐量，因?yàn)樗鼈儨p少了對(duì)特定資源的爭(zhēng)用。盡管完全可以手動(dòng)編寫(xiě)限制代碼，但使用 Semaphore類(lèi)可以更輕松地完成此任務(wù)，它將幫您執(zhí)行限制。

常用API

public void acquire() // 獲取許可。
public void acquireUninterruptibly()
public boolean tryAcquire()
public boolean tryAcquire(long timeout, TimeUnit unit)
public void release() // 釋放許可。該方法一般調(diào)用于finally塊中。

例：10 個(gè)線(xiàn)程都在運(yùn)行，可以對(duì)運(yùn)行SemaphoreApp的Java進(jìn)程執(zhí)行jstack來(lái)驗(yàn)證，只有3個(gè)線(xiàn)程是活躍的。在一個(gè)信號(hào)計(jì)數(shù)器釋放之前，其他7個(gè)線(xiàn)程都處于空閑狀態(tài)。

import java.util.Random;
import java.util.concurrent.Semaphore; 
public class SemaphoreApp { 
public static void main(String[] args) { // 匿名Runnable實(shí)例。定義線(xiàn)程運(yùn)行程序。
        Runnable limitedCall = new Runnable() { final Random rand = new Random(); final Semaphore available = new Semaphore(3);     // 最多可以發(fā)出3個(gè)"許可"
            int count = 0; public void run() { int time = rand.nextInt(15); int num = count++; try {
                    available.acquire(); // 當(dāng)前線(xiàn)程獲取"許可"。若沒(méi)有獲取許可，則等待于此。
                    System.out.println("Executing " + "long-running action for " + time + " seconds... #" + num); 
                    Thread.sleep(time * 1000); 
                    System.out.println("Done with #" + num + "!"); 
                } catch (InterruptedException intEx) { 
                    intEx.printStackTrace(); 
                } finally {
                    available.release(); // 當(dāng)前線(xiàn)程釋放"許可"
 }
            } 
        }; for (int i = 0; i < 10; i++) { new Thread(limitedCall).start(); 
        }
}

例：停車(chē)示例。停車(chē)場(chǎng)只有10個(gè)車(chē)位，現(xiàn)在有30輛車(chē)去停車(chē)。當(dāng)車(chē)位滿(mǎn)時(shí)出來(lái)一輛車(chē)才能有一輛車(chē)進(jìn)入停車(chē)。

import java.util.concurrent.Semaphore; public class Car implements Runnable { private final Semaphore parkingSlot; private int carNo; public Car(Semaphore parkingSlot, int carNo) { this.parkingSlot = parkingSlot; this.carNo = carNo;
    } public void run() { try {
            parkingSlot.acquire(); // 車(chē)嘗試獲取"車(chē)位"
 parking();
            sleep(300);
            leaving();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            parkingSlot.release(); // 釋放"車(chē)位"
 }
    } private void parking() {
        System.out.println(String.format("%d號(hào)車(chē)泊車(chē)", carNo));
    } private void leaving() {
        System.out.println(String.format("%d號(hào)車(chē)離開(kāi)車(chē)位", carNo));
    } private static void sleep(long millis) { try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

} // --------------------------------------------------------------------------
import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; public class ParkingCars { private static final int NUMBER_OF_CARS = 30; private static final int NUMBER_OF_PARKING_SLOT = 10; public static void main(String[] args) {
        Semaphore parkingSlot = new Semaphore(NUMBER_OF_PARKING_SLOT, true);    // "車(chē)位",采用FIFO,設(shè)置true。
 ExecutorService service = Executors.newCachedThreadPool();    // 創(chuàng)建線(xiàn)程池。模擬30輛車(chē)"停車(chē)"。
        for (int carNo = 1; carNo <= NUMBER_OF_CARS; carNo++) {
            service.execute(new Car(parkingSlot, carNo));
        }
        sleep(3000);
        service.shutdown(); // 關(guān)閉線(xiàn)程池。 // 輸出剩余可以用的資源數(shù)。
        System.out.println(parkingSlot.availablePermits() + " 個(gè)停車(chē)位可以用!");
    } private static void sleep(long millis) { try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

Exchanger 交換器

Exchanger用于實(shí)現(xiàn)線(xiàn)程間的數(shù)據(jù)交換。Exchanger提供一個(gè)同步點(diǎn)，在同步點(diǎn)上，兩個(gè)線(xiàn)程使用exchange方法交換彼此數(shù)據(jù)。如果第一個(gè)線(xiàn)程先執(zhí)行exchange方法，則它會(huì)等待第二個(gè)線(xiàn)程執(zhí)行exchange方法。當(dāng)兩個(gè)線(xiàn)程同時(shí)到達(dá)同步點(diǎn)時(shí)，這兩個(gè)線(xiàn)程即可以交換數(shù)據(jù)。交換完畢后，各自進(jìn)行以后的程序流程。當(dāng)兩個(gè)線(xiàn)程通過(guò)Exchanger交換數(shù)據(jù)的時(shí)候，這個(gè)交換對(duì)于兩個(gè)線(xiàn)程來(lái)說(shuō)是線(xiàn)程安全的。

exchange()方法將本線(xiàn)程的數(shù)據(jù)作為參數(shù)，傳遞給伙伴線(xiàn)程，并且該方法返回伙伴線(xiàn)程提供的數(shù)據(jù)。

當(dāng)在運(yùn)行不對(duì)稱(chēng)的活動(dòng)時(shí)Exchanger很有用，比如當(dāng)一個(gè)線(xiàn)程填充了buffer，另一個(gè)線(xiàn)程從buffer中消費(fèi)數(shù)據(jù)時(shí)，這兩個(gè)線(xiàn)程可以用Exchanger來(lái)交換數(shù)據(jù)。

Exchanger<V>類(lèi)的API

構(gòu)造器

Exchanger() 創(chuàng)建一個(gè)新的 Exchanger。

方法

V exchange(V x) 等待另一個(gè)線(xiàn)程到達(dá)此交換點(diǎn)（除非當(dāng)前線(xiàn)程被中斷），然后將給定的對(duì)象傳送給該線(xiàn)程，并接收該線(xiàn)程的對(duì)象。

V exchange(V x, long timeout, TimeUnit unit) 等待另一個(gè)線(xiàn)程到達(dá)此交換點(diǎn)（除非當(dāng)前線(xiàn)程被中斷，或者超出了指定的等待時(shí)間），然后將給定的對(duì)象傳送給該線(xiàn)程，同時(shí)接收該線(xiàn)程的對(duì)象。

例：以下這個(gè)程序demo要做的事情就是生產(chǎn)者在交換前生產(chǎn)5個(gè)"生產(chǎn)者"，然后再與消費(fèi)者交換5個(gè)數(shù)據(jù)，然后再生產(chǎn)5個(gè)"交換后生產(chǎn)者"，而消費(fèi)者要在交換前消費(fèi)5個(gè)"消費(fèi)者"，然后再與生產(chǎn)者交換5個(gè)數(shù)據(jù)，然后再消費(fèi)5個(gè)"交換后消費(fèi)者"。importjava.util.ArrayList;

import java.util.Iterator; import java.util.List; import java.util.concurrent.Exchanger; /** * 兩個(gè)線(xiàn)程間的數(shù)據(jù)交換 */
public class ExchangerDemo { private static final Exchanger<List<String>> ex = new Exchanger<List<String>>(); private static void sleep(long millis){ try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    } /** * 內(nèi)部類(lèi)，數(shù)據(jù)生成者 */
    class DataProducer implements Runnable { private List<String> list = new ArrayList<String>(); public void run() {
            System.out.println("生產(chǎn)者開(kāi)始生產(chǎn)數(shù)據(jù)"); for (int i = 1; i <= 5; i++) {
                System.out.println("生產(chǎn)了第" + i + "個(gè)數(shù)據(jù)，耗時(shí)1秒");
                list.add("生產(chǎn)者" + i);
                sleep(1000);
            }
            System.out.println("生產(chǎn)數(shù)據(jù)結(jié)束");
            System.out.println("開(kāi)始與消費(fèi)者交換數(shù)據(jù)"); try { //將數(shù)據(jù)準(zhǔn)備用于交換，并返回消費(fèi)者的數(shù)據(jù)
                list = (List<String>) ex.exchange(list);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("結(jié)束與消費(fèi)者交換數(shù)據(jù)");
            System.out.println("生產(chǎn)者與消費(fèi)者交換數(shù)據(jù)后，再生產(chǎn)數(shù)據(jù)"); for (int i = 6; i < 10; i++) {
                System.out.println("交換后生產(chǎn)了第" + i + "個(gè)數(shù)據(jù)，耗時(shí)1秒");
                list.add("交換后生產(chǎn)者" + i);
                sleep(1000);
            }

            System.out.println("遍歷生產(chǎn)者交換后的數(shù)據(jù)"); for (Iterator<String> iterator = list.iterator(); iterator.hasNext();) {
                System.out.println(iterator.next());
            }
        }

    } /** * 內(nèi)部類(lèi)，數(shù)據(jù)消費(fèi)者 */
    class DataConsumer implements Runnable { private List<String> list = new ArrayList<String>(); public void run() {
            System.out.println("消費(fèi)者開(kāi)始消費(fèi)數(shù)據(jù)"); for (int i = 1; i <= 5; i++) {
                System.out.println("消費(fèi)了第" + i + "個(gè)數(shù)據(jù)"); // 消費(fèi)者產(chǎn)生數(shù)據(jù)，后面交換的時(shí)候給生產(chǎn)者
                list.add("消費(fèi)者" + i);
             }

            System.out.println("消費(fèi)數(shù)據(jù)結(jié)束");
            System.out.println("開(kāi)始與生產(chǎn)者交換數(shù)據(jù)"); try { // 進(jìn)行數(shù)據(jù)交換，返回生產(chǎn)者的數(shù)據(jù)
                list = (List<String>) ex.exchange(list);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("消費(fèi)者與生產(chǎn)者交換數(shù)據(jù)后，再消費(fèi)數(shù)據(jù)"); for (int i = 6; i < 10; i++) {
                System.out.println("交換后消費(fèi)了第" + i + "個(gè)數(shù)據(jù)");
                list.add("交換后消費(fèi)者" + i);
                sleep(1000);
            }
            sleep(1000);
            System.out.println("開(kāi)始遍歷消費(fèi)者交換后的數(shù)據(jù)"); for (Iterator<String> iterator = list.iterator(); iterator.hasNext();) {
                System.out.println(iterator.next());
            }
        }
    } // 主方法
    public static void main(String args[]) {
        ExchangerDemo et = new ExchangerDemo(); new Thread(et.new DataProducer()).start(); new Thread(et.new DataConsumer()).start();
    }
}