線程通訊指的是多個(gè)線程之間通過(guò)共享內(nèi)存或消息傳遞等方式來(lái)協(xié)調(diào)和同步它們的執(zhí)行。在多線程編程中，通常會(huì)出現(xiàn)多個(gè)線程需要共同完成某個(gè)任務(wù)的情況，這時(shí)就需要線程之間進(jìn)行通訊，以保證任務(wù)能夠順利地執(zhí)行。

線程通訊的實(shí)現(xiàn)方式主要有以下兩種：

• 共享內(nèi)存：多個(gè)線程可以訪問(wèn)同一個(gè)共享內(nèi)存區(qū)域，通過(guò)讀取和寫入內(nèi)存中的數(shù)據(jù)來(lái)進(jìn)行通訊和同步。

• 消息傳遞：多個(gè)線程之間通過(guò)消息隊(duì)列、管道、信號(hào)量等機(jī)制來(lái)傳遞信息和同步狀態(tài)。

常見場(chǎng)景

線程通訊的常見場(chǎng)景有以下幾個(gè)：

• 多個(gè)線程共同完成某個(gè)任務(wù)：例如一個(gè)爬蟲程序需要多個(gè)線程同時(shí)抓取不同的網(wǎng)頁(yè)，然后將抓取結(jié)果合并保存到數(shù)據(jù)庫(kù)中。這時(shí)需要線程通訊來(lái)協(xié)調(diào)各個(gè)線程的執(zhí)行順序和共享數(shù)據(jù)。

• 避免資源沖突：多個(gè)線程訪問(wèn)共享資源時(shí)可能會(huì)引發(fā)競(jìng)爭(zhēng)條件，例如多個(gè)線程同時(shí)讀寫一個(gè)文件或數(shù)據(jù)庫(kù)。這時(shí)需要線程通訊來(lái)同步線程之間的數(shù)據(jù)訪問(wèn)，避免資源沖突。

• 保證順序執(zhí)行：在某些情況下，需要保證多個(gè)線程按照一定的順序執(zhí)行，例如一個(gè)多線程排序算法。這時(shí)需要線程通訊來(lái)協(xié)調(diào)各個(gè)線程的執(zhí)行順序。

• 線程之間的互斥和同步：有些場(chǎng)景需要確保只有一個(gè)線程能夠訪問(wèn)某個(gè)共享資源，例如一個(gè)計(jì)數(shù)器。這時(shí)需要使用線程通訊機(jī)制來(lái)實(shí)現(xiàn)線程之間的互斥和同步。

實(shí)現(xiàn)方法

線程通訊的實(shí)現(xiàn)方法有以下幾種：

• 等待和通知機(jī)制：使用 Object 類的 wait() 和 notify() 方法來(lái)實(shí)現(xiàn)線程之間的通訊。當(dāng)一個(gè)線程需要等待另一個(gè)線程執(zhí)行完某個(gè)操作時(shí)，它可以調(diào)用 wait() 方法使自己進(jìn)入等待狀態(tài)，同時(shí)釋放占有的鎖，等待其他線程調(diào)用 notify() 或 notifyAll() 方法來(lái)喚醒它。被喚醒的線程會(huì)重新嘗試獲取鎖并繼續(xù)執(zhí)行。

• 信號(hào)量機(jī)制：使用 Java 中的 Semaphore 類來(lái)實(shí)現(xiàn)線程之間的同步和互斥。Semaphore 是一個(gè)計(jì)數(shù)器，用來(lái)控制同時(shí)訪問(wèn)某個(gè)資源的線程數(shù)。當(dāng)某個(gè)線程需要訪問(wèn)共享資源時(shí)，它必須先從 Semaphore 中獲取一個(gè)許可證，如果已經(jīng)沒(méi)有許可證可用，線程就會(huì)被阻塞，直到其他線程釋放了許可證。

• 柵欄機(jī)制：使用 Java 中的 CyclicBarrier 類來(lái)實(shí)現(xiàn)多個(gè)線程之間的同步，它允許多個(gè)線程在指定的屏障處等待，并在所有線程都達(dá)到屏障時(shí)繼續(xù)執(zhí)行。

• 鎖機(jī)制：使用 Java 中的 Lock 接口和 Condition 接口來(lái)實(shí)現(xiàn)線程之間的同步和互斥。Lock 是一種更高級(jí)的互斥機(jī)制，它允許多個(gè)條件變量（Condition）并支持在同一個(gè)鎖上等待和喚醒。

具體代碼實(shí)現(xiàn)

等待通知實(shí)現(xiàn)

以下是一個(gè)簡(jiǎn)單的 wait() 和 notify() 方法的等待通知示例：

public class WaitNotifyDemo {
public static void main(String[] args) {
        Object lock = new Object();
        ThreadA threadA = new ThreadA(lock);
        ThreadB threadB = new ThreadB(lock);
        threadA.start();
        threadB.start();
}
static class ThreadA extends Thread {
        private Object lock;
        public ThreadA(Object lock) {
            this.lock = lock;
        }
        public void run() {
            synchronized (lock) {
                System.out.println("ThreadA start...");
                try {
                    lock.wait(); // 線程A等待
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("ThreadA end...");
            }
        }
    }
static class ThreadB extends Thread {
        private Object lock;
        public ThreadB(Object lock) {
            this.lock = lock;
        }
        public void run() {
            synchronized (lock) {
                System.out.println("ThreadB start...");
                lock.notify(); // 喚醒線程A
                System.out.println("ThreadB end...");
            }
        }
    }
   }

在這個(gè)示例中，定義了一個(gè)共享對(duì)象 lock，ThreadA 線程先獲取 lock 鎖，并調(diào)用 lock.wait() 方法進(jìn)入等待狀態(tài)。ThreadB 線程在獲取 lock 鎖之后，調(diào)用 lock.notify() 方法喚醒 ThreadA 線程，然后 ThreadB 線程執(zhí)行完畢。

運(yùn)行以上程序的執(zhí)行結(jié)果如下：

ThreadA start...

ThreadB start...

ThreadB end...

ThreadA end...

信號(hào)量實(shí)現(xiàn)

在 Java 中使用 Semaphore 實(shí)現(xiàn)信號(hào)量，Semaphore 是一個(gè)計(jì)數(shù)器，用來(lái)控制同時(shí)訪問(wèn)某個(gè)資源的線程數(shù)。當(dāng)某個(gè)線程需要訪問(wèn)共享資源時(shí)，它必須先從 Semaphore 中獲取一個(gè)許可證，如果已經(jīng)沒(méi)有許可證可用，線程就會(huì)被阻塞，直到其他線程釋放了許可證。它的示例代碼如下：

import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(2);
        for (int i = 0; i < 5; i++) {
            new Thread(new Worker(i, semaphore)).start();
        }
    }
    static class Worker implements Runnable {
        private int id;
        private Semaphore semaphore;
        public Worker(int id, Semaphore semaphore) {
            this.id = id;
            this.semaphore = semaphore;
        }
        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("Worker " + id + " acquired permit.");
                Thread.sleep(1000);
                System.out.println("Worker " + id + " released permit.");
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

在這個(gè)示例中，創(chuàng)建了一個(gè) Semaphore 對(duì)象，并且設(shè)置了許可數(shù)為 2。然后創(chuàng)建了 5 個(gè) Worker 線程，每個(gè) Worker 線程需要獲取 Semaphore 的許可才能執(zhí)行任務(wù)。每個(gè) Worker 線程在執(zhí)行任務(wù)之前先調(diào)用 semaphore.acquire() 方法獲取許可，如果沒(méi)有許可則會(huì)阻塞，直到 Semaphore 釋放許可。執(zhí)行完任務(wù)之后調(diào)用 semaphore.release() 方法釋放許可。 運(yùn)行以上程序的執(zhí)行結(jié)果如下：

Worker 0 acquired permit.

Worker 1 acquired permit.

Worker 1 released permit.

Worker 0 released permit.

Worker 2 acquired permit.

Worker 3 acquired permit.

Worker 2 released permit.

Worker 4 acquired permit.

Worker 3 released permit.

Worker 4 released permit.

柵欄實(shí)現(xiàn)

在 Java 中，可以使用 CyclicBarrier 或 CountDownLatch 來(lái)實(shí)現(xiàn)線程的同步，它們兩個(gè)使用類似，接下來(lái)我們就是 CyclicBarrier 來(lái)演示一下線程的同步，CyclicBarrier 的示例代碼如下：

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new Runnable() {
            @Override
            public void run() {
                System.out.println("All threads have reached the barrier.");
            }
        });
        for (int i = 1; i <= 3; i++) {
            new Thread(new Worker(i, cyclicBarrier)).start();
        }
    }
    static class Worker implements Runnable {
        private int id;
        private CyclicBarrier cyclicBarrier;
        public Worker(int id, CyclicBarrier cyclicBarrier) {
            this.id = id;
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                System.out.println("Worker " + id + " is working.");
                Thread.sleep((long) (Math.random() * 2000));
                System.out.println("Worker " + id + " has reached the barrier.");
                cyclicBarrier.await();
                System.out.println("Worker " + id + " is continuing the work.");
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}

在這個(gè)示例中，創(chuàng)建了一個(gè) CyclicBarrier 對(duì)象，并且設(shè)置了參與線程數(shù)為 3。然后創(chuàng)建了 3 個(gè) Worker 線程，每個(gè) Worker 線程會(huì)先執(zhí)行一些任務(wù)，然后等待其他線程到達(dá) Barrier。所有線程都到達(dá) Barrier 之后，Barrier 會(huì)釋放所有線程并執(zhí)行設(shè)置的 Runnable 任務(wù)。 運(yùn)行以上程序的執(zhí)行結(jié)果如下：

Worker 2 is working.

Worker 3 is working.

Worker 1 is working.

Worker 3 has reached the barrier.

Worker 1 has reached the barrier.

Worker 2 has reached the barrier.

All threads have reached the barrier.

Worker 2 is continuing the work.

Worker 3 is continuing the work.

Worker 1 is continuing the work.

從以上執(zhí)行結(jié)果可以看出，CyclicBarrier 保證了所有 Worker 線程都到達(dá) Barrier 之后才能繼續(xù)執(zhí)行后面的任務(wù)，這樣可以保證線程之間的同步和協(xié)作。在本示例中，所有線程都在 Barrier 處等待了一段時(shí)間，等所有線程都到達(dá) Barrier 之后才繼續(xù)執(zhí)行后面的任務(wù)。

鎖機(jī)制實(shí)現(xiàn)

以下是一個(gè)使用 Condition 的示例：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ConditionDemo {
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private volatile boolean flag = false;
    public static void main(String[] args) {
        ConditionDemo demo = new ConditionDemo();
        new Thread(demo::waitCondition).start();
        new Thread(demo::signalCondition).start();
    }
    private void waitCondition() {
        lock.lock();
        try {
            while (!flag) {
                System.out.println(Thread.currentThread().getName() + " is waiting for signal.");
                condition.await();
            }
            System.out.println(Thread.currentThread().getName() + " received signal.");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    private void signalCondition() {
        lock.lock();
        try {
            Thread.sleep(3000); // 模擬等待一段時(shí)間后發(fā)送信號(hào)
            flag = true;
            System.out.println(Thread.currentThread().getName() + " sends signal.");
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

在這個(gè)示例中，創(chuàng)建了一個(gè) Condition 對(duì)象和一個(gè) Lock 對(duì)象，然后創(chuàng)建了兩個(gè)線程，一個(gè)線程等待 Condition 信號(hào)，另一個(gè)線程發(fā)送 Condition 信號(hào)。

等待線程在獲得鎖后，判斷標(biāo)志位是否為 true，如果為 false，則等待 Condition 信號(hào)；如果為 true，則繼續(xù)執(zhí)行后面的任務(wù)。

發(fā)送線程在獲得鎖后，等待一段時(shí)間后，將標(biāo)志位設(shè)置為 true，并且發(fā)送 Condition 信號(hào)。 運(yùn)行以上程序的執(zhí)行結(jié)果如下：

Thread-0 is waiting for signal.

Thread-1 sends signal.

Thread-0 received signal.

從上面執(zhí)行結(jié)果可以看出，等待線程在等待 Condition 信號(hào)的時(shí)候被阻塞，直到發(fā)送線程發(fā)送了 Condition 信號(hào)，等待線程才繼續(xù)執(zhí)行后面的任務(wù)。Condition 對(duì)象提供了一種更加靈活的線程通信方式，可以精確地控制線程的等待和喚醒。

小結(jié)

線程通訊指的是多個(gè)線程之間通過(guò)共享內(nèi)存或消息傳遞等方式來(lái)協(xié)調(diào)和同步它們的執(zhí)行，它的實(shí)現(xiàn)方法有很多：比如 wait() 和 notify() 的等待和通知機(jī)制、Semaphore 信號(hào)量機(jī)制、CyclicBarrier 柵欄機(jī)制，以及 Condition 的鎖機(jī)制等。

以上就是Java線程通訊的實(shí)現(xiàn)方式總結(jié)的詳細(xì)內(nèi)容，更多關(guān)于Java線程通訊的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論