1 Exchanger 介紹

前面分別介紹了CyclicBarrier、CountDownLatch、Semaphore，現(xiàn)在介紹并發(fā)工具類中的最后一個(gè)Exchange。
Exchanger 是一個(gè)用于線程間協(xié)作的工具類，Exchanger用于進(jìn)行線程間的數(shù)據(jù)交換，它提供一個(gè)同步點(diǎn)，在這個(gè)同步點(diǎn)，兩個(gè)線程可以交換彼此的數(shù)據(jù)。這兩個(gè)線程通過exchange 方法交換數(shù)據(jù)，如果第一個(gè)線程先執(zhí)行exchange 方法，它會(huì)一直等待第二個(gè)線程也執(zhí)行exchange 方法，當(dāng)兩個(gè)線程都到達(dá)同步點(diǎn)時(shí)，這兩個(gè)線程就可以交換數(shù)據(jù)。

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.

在以上的描述中，有幾個(gè)要點(diǎn)：

• 此類提供對(duì)外的操作是同步的；
• 用于成對(duì)出現(xiàn)的線程之間交換數(shù)據(jù)；
• 可以視作雙向的同步隊(duì)列；
• 可應(yīng)用于基因算法、流水線設(shè)計(jì)等場(chǎng)景。
• 接著看api文檔，這個(gè)類提供對(duì)外的接口非常簡潔，一個(gè)無參構(gòu)造函數(shù)，兩個(gè)重載的范型exchange方法：
  

public V exchange(V x) throws InterruptedException
public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException

2 Exchanger 實(shí)例

public class ExchangerTest {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        final Exchanger exchanger = new Exchanger();
        executor.execute(new Runnable() {
            String data = "data1";

            @Override
            public void run() {
                doExchangeWork(data, exchanger);
            }
        });

        executor.execute(new Runnable() {
            String data = "data2";

            @Override
            public void run() {
                doExchangeWork(data, exchanger);
            }
        });
        executor.shutdown();
    }

    private static void doExchangeWork(String data, Exchanger exchanger) {
        try {
            System.out.println(Thread.currentThread().getName() + "正在把數(shù)據(jù) " + data + " 交換出去");
            Thread.sleep((long) (Math.random() * 1000));

            String exchangeData = (String) exchanger.exchange(data);
            System.out.println(Thread.currentThread().getName() + "交換得到數(shù)據(jù)  " + exchangeData);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

pool-1-thread-1正在把數(shù)據(jù) data1 交換出去
pool-1-thread-2正在把數(shù)據(jù) data2 交換出去
pool-1-thread-2交換得到數(shù)據(jù)  data1
pool-1-thread-1交換得到數(shù)據(jù)  data2

當(dāng)線程A調(diào)用Exchange對(duì)象的exchange()方法后，他會(huì)陷入阻塞狀態(tài)，直到線程B也調(diào)用了exchange()方法，然后以線程安全的方式交換數(shù)據(jù)，之后線程A和B繼續(xù)運(yùn)行。

exchange等待超時(shí)

public class ExchangerTest {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        final Exchanger exchanger = new Exchanger();
        executor.execute(new Runnable() {
            String data = "data1";

            @Override
            public void run() {
                doExchangeWork(data, exchanger);
            }
        });

        executor.execute(new Runnable() {
            String data = "data2";

            @Override
            public void run() {
                try {
                    Thread.sleep((long) (3000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                doExchangeWork(data, exchanger);
            }
        });
        executor.shutdown();
    }

    private static void doExchangeWork(String data, Exchanger exchanger) {
        try {
            System.out.println(Thread.currentThread().getName() + "正在把數(shù)據(jù) " + data + " 交換出去");

            //遠(yuǎn)小于3秒拋出異常
            String exchangeData = (String) exchanger.exchange(data,1, TimeUnit.SECONDS);
            System.out.println(Thread.currentThread().getName() + "交換得到數(shù)據(jù)  " + exchangeData);
        } catch ( TimeoutException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

pool-1-thread-1正在把數(shù)據(jù) data1 交換出去
java.util.concurrent.TimeoutException
    at java.util.concurrent.Exchanger.exchange(Exchanger.java:626)
    at ExchangerTest.doExchangeWork(ExchangerTest.java:37)
    at ExchangerTest.access$000(ExchangerTest.java:3)
    at ExchangerTest$1.run(ExchangerTest.java:12)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
pool-1-thread-2正在把數(shù)據(jù) data2 交換出去
java.util.concurrent.TimeoutException
    at java.util.concurrent.Exchanger.exchange(Exchanger.java:626)
    at ExchangerTest.doExchangeWork(ExchangerTest.java:37)
    at ExchangerTest.access$000(ExchangerTest.java:3)
    at ExchangerTest$2.run(ExchangerTest.java:26)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

實(shí)戰(zhàn)場(chǎng)景：

設(shè)計(jì)一個(gè)定時(shí)任務(wù)，每日凌晨執(zhí)行。在定時(shí)任務(wù)中啟動(dòng)兩個(gè)線程，一個(gè)線程負(fù)責(zé)對(duì)業(yè)務(wù)明細(xì)表（xxx_info）進(jìn)行查詢統(tǒng)計(jì)，把統(tǒng)計(jì)的結(jié)果放置在內(nèi)存緩沖區(qū)，另一個(gè)線程負(fù)責(zé)讀取緩沖區(qū)中的統(tǒng)計(jì)結(jié)果并插入到業(yè)務(wù)統(tǒng)計(jì)表（xxx_statistics）中。
親，這樣的場(chǎng)景是不是聽起來很有感覺？沒錯(cuò)!兩個(gè)線程在內(nèi)存中批量交換數(shù)據(jù)，這個(gè)事情我們可以使用Exchanger去做！

3 實(shí)現(xiàn)原理

Exchanger（交換者）是一個(gè)用于線程間協(xié)作的工具類。Exchanger用于進(jìn)行線程間的數(shù)據(jù)交換。它提供一個(gè)同步點(diǎn)，在這個(gè)同步點(diǎn)兩個(gè)線程可以交換彼此的數(shù)據(jù)。這兩個(gè)線程通過exchange方法交換數(shù)據(jù)， 如果第一個(gè)線程先執(zhí)行exchange方法，它會(huì)一直等待第二個(gè)線程也執(zhí)行exchange，當(dāng)兩個(gè)線程都到達(dá)同步點(diǎn)時(shí)，這兩個(gè)線程就可以交換數(shù)據(jù)，將本線程生產(chǎn)出來的數(shù)據(jù)傳遞給對(duì)方。因此使用Exchanger的重點(diǎn)是成對(duì)的線程使用exchange()方法，當(dāng)有一對(duì)線程達(dá)到了同步點(diǎn)，就會(huì)進(jìn)行交換數(shù)據(jù)。因此該工具類的線程對(duì)象是成對(duì)的。
Exchanger類提供了兩個(gè)方法，String exchange(V x):用于交換，啟動(dòng)交換并等待另一個(gè)線程調(diào)用exchange；String exchange(V x,long timeout,TimeUnit unit)：用于交換，啟動(dòng)交換并等待另一個(gè)線程調(diào)用exchange，并且設(shè)置最大等待時(shí)間，當(dāng)?shù)却龝r(shí)間超過timeout便停止等待。

到此這篇關(guān)于Java多線程之同步工具類Exchanger的文章就介紹到這了,更多相關(guān)Java多線程 Exchanger內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論