Javaでスレッド間の通信を実装するためにwaitとnotifyを使用する方法-＆＃＆チュートリアル-php.cn

1. スレッド通信が必要な理由

スレッドは同時に実行され、これはスレッドのランダムな実行のように見えますが、実際のアプリケーションではスレッドを使用します。通信実行順序が必要なので、スレッド通信を使用する必要があります。

スレッド通信スレッドの実行順序を解決するために優先順位を使用してみてはいかがでしょうか?

全体の優先度は、スレッド PCB 内の優先度情報とスレッドの待機時間によって決定されるため、一般的な開発では、スレッドの実行順序を示すために優先度に依存することはありません。

参照以下のようなシーン: 生産者と消費者モデルを説明するためのパン屋の例

生産者と消費者に対応する、パン屋と顧客がいるパン屋があります。パン屋にはパンを保管するための在庫があります。在庫がいっぱいになると、生産されなくなります。同時に、消費者はパンも購入します。パンの在庫が売り切れると、消費者は引き続き購入する前に、新しいパンが生産されるまで待たなければなりません

分析: いつ生産を停止し、いつ消費を停止するかについて、生産と消費の情報を正確に伝えるためにスレッド通信を適用する必要があります

2. 待機メソッドと通知メソッド

wait() : 現在のスレッドによって保持されているオブジェクトを許可します。ロックを解放して待機します。

wait(long timeout): 対応するパラメータは、スレッドが待機する時間です。

notify(): スレッドを起動します。同じオブジェクトを使用して wait を呼び出し、待機中のスレッドに入り、再度オブジェクトロックを競合します。

notifyAll(): 複数のスレッドが待機している場合、notifyAll はすべてのスレッドを起動し、notify はランダムに 1 つを起動します

注:

これらのメソッドはすべて、オブジェクトクラスのメソッドに属します。

同期された同期コードブロック/同期メソッドで使用する必要があります

どのオブジェクトがロックされているかは、wait と notify に使用されるオブジェクトです。

notify 直後に起動せず、同期が完了するまで待ってから起動します。

1 .wait() メソッド

wait メソッドの呼び出し後:

現在のコードを実行しているスレッドを待機させます (スレッドは待機キューに入れられます)

現在のロックを解放します

特定の条件が満たされると起動し、再度ロックの取得を試みます

待機終了の条件:

他のスレッドがオブジェクトのnotifyメソッドを呼び出す

wait待機時間タイムアウト(待ち時間を指定するタイムアウトパラメータ)

他のスレッドが割り込みメソッドを呼び出すとwaitがInterruptedExceptionをスローする

2. Notice() メソッド

##パラメータなしで wait メソッドを使用する場合は、notify メソッドを使用してスレッドをウェイクアップして待機する必要があります

このメソッドはウェイクアップするためのものですオブジェクトのオブジェクトロックを待機しているスレッドを起動して、オブジェクトのオブジェクトロックを再取得できるようにします。

待機中のスレッドが複数ある場合、スレッドスケジューラは待機ステータススレッドをランダムに選択します (最初のスレッドはありません)。 -come, first-served)

notify() メソッドの後、現在のスレッドはオブジェクトロックをすぐに解放せず、notify() メソッドを実行しているスレッドがプログラムの実行を終了するまで待機します。、同期コードを終了しますオブジェクトのロックは、ブロックの後でのみ解放されます

3.notifyAll() メソッド

このメソッドは、起動時を除き、notify() メソッドと同じ効果があります。起動すると、待機中のすべてのスレッドが起動されます

notify() メソッドはランダムにスレッドを起動するだけです

3. wait と Notice を使用してベーカリービジネスを実装します

前提条件:

はい 2 人のパン屋がいます。パン屋は一度に 2 つのパンを作ることができます。

倉庫には 100 のパンを保管できます。

消費者は 10 人いて、各消費者は 1 つずつ購入します

注:

消費と生産は同時に並行して実行され、1 つの生産と 1 つの消費ではありません。 #実装コード:

public class Bakery {
    private static int total;//库存
  
    public static void main(String[] args) {
        Producer producer = new Producer();
        for(int i = 0;i < 2;i++){
            new Thread(producer,"面包师傅-"+(i-1)).start();
        }
        Consumer consumer = new Consumer();
        for(int i = 0;i < 10;i++){
            new Thread(consumer,"消费者-"+(i-1)).start();
        }
    }
    private static class Producer implements Runnable{
        private int num = 3; //生产者每次生产三个面包
        @Override
        public void run() {
            try {
                while(true){ //一直生产
                    synchronized (Bakery.class){
                        while((total+num)>100){ //仓库满了，生产者等待
                            Bakery.class.wait();
                        }
                        //等待解除
                        total += num;
                        System.out.println(Thread.currentThread().getName()+"生产面包，库存："+total);
                        Thread.sleep(500);
                        Bakery.class.notifyAll(); //唤醒生产
                    }
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    private static class Consumer implements Runnable{
        private int num = 1; //消费者每次消费1个面包
        @Override
        public void run() {
            try {
                while(true){ //一直消费
                    synchronized (Bakery.class){
                        while((total-num)<0){ //仓库空了，消费者等待
                            Bakery.class.wait();
                        }
                        //解除消费者等待
                        total -= num;
                        System.out.println(Thread.currentThread().getName()+"消费面包，库存："+total);
                        Thread.sleep(500);
                        Bakery.class.notifyAll(); //唤醒消费
                    }
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

ログイン後にコピー

部分的な印刷結果:

4. ブロッキングキュー

Javaでスレッド間の通信を実装するためにwaitとnotifyを使用する方法ブロッキングキューは特殊なキューです。これも「先入れ先出し」の原則に従っており、スレッドセーフなキュー構造です。

特徴: 典型的な生産者/消費者モデルで、一般にタスクの分離とピーク削減に使用されます

キューがいっぱいの場合、キューはブロックされて待機します (実稼働)、他のスレッドがキューから要素を取得するまで

キューが空の場合、キューはブロックされて待機します (消費)、他のスレッドが要素を挿入するまで待機します

1. プロデューサーコンシューマーモデル

プロデューサー/コンシューマーモデルは、コンテナーを使用して、プロデューサーとコンシューマー間の強い結合の問題を解決します。

プロデューサーとコンシューマーは、次のことを行います。相互に直接通信するのではなく、ブロッキングを介して通信しますキューは通信に使用されるため、プロデューサーがデータを生成した後、コンシューマーがデータを処理するのを待って、それをブロッキングキューに直接スローします。コンシューマーはプロデューサーに要求しません。

ブロッキングキューは、バッファ内でプロデューサーとコンシューマの処理能力のバランスをとります。

ブロッキングキューは、プロデューサーとコンシューマを分離することもできます

上記のベーカリービジネスの実装は生産者消費者モデルです。例

2. 標準ライブラリのブロッキングキュー

ブロッキングキューは Java 標準ライブラリに組み込まれています。一部のプログラムではブロッキングキューを使用する必要がありますが、標準ライブラリではブロッキングキューを直接使用する必要があります。

BlockingQueue 是一个接口. 真正实现的类是 LinkedBlockingQueue

put 方法用于阻塞式的入队列, take 用于阻塞式的出队列

BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性

BlockingDeque<String> queue = new LinkedBlockingDeque<>();
queue.put("hello");
//如果队列为空，直接出出队列就会阻塞
String ret = queue.take();
System.out.println(ret);

ログイン後にコピー

3. 阻塞队列的模拟实现

这里使用数组实现一个循环队列来模拟阻塞队列

当队列为空的时候，就不能取元素了，就进入wait等待，当有元素存放时，唤醒

当队列为满的时候，就不能存元素了，就进入wait等待，当铀元素取出时，唤醒

实现代码：

public class MyBlockingQueue {
    //使用数组实现一个循环队列，队列里面存放的是线程要执行的任务
    private Runnable[] tasks;
    //队列中任务的数量，根据数量来判断是否可以存取
    private int count;
    private int putIndex; //存放任务位置
    private int takeIndex; //取出任务位置
  
    //有参的构造方法，表示队列容量
    public MyBlockingQueue(int size){
        tasks = new Runnable[size];
    }
  
    //存任务
    public void put(Runnable task){
        try {
            synchronized (MyBlockingQueue.class){
                //如果队列容量满了，则存任务等待
                while(count == tasks.length){
                    MyBlockingQueue.class.wait();
                }
                tasks[putIndex] = task; //将任务放入数组
                putIndex = (putIndex+1) % tasks.length; //更新存任务位置
                count++; //更新存放数量
                MyBlockingQueue.class.notifyAll(); //唤醒存任务
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
  
    //取任务
    public Runnable take(){
        try {
            synchronized (MyBlockingQueue.class){
                //如果队列任务为空，则取任务等待
                while(count==0){
                    MyBlockingQueue.class.wait();
                }
                //取任务
                Runnable task = tasks[takeIndex];
                takeIndex = (takeIndex+1) % tasks.length; //更新取任务位置
                count--; //更新存放数量
                MyBlockingQueue.class.notifyAll(); //唤醒取任务
                return task;
            }
        } catch (InterruptedException e) {
           throw new RuntimeException("存放任务出错",e);
        }
    }
}