Java BlockingQueueの例

今日はJavaのBlockingQueueについて見ていきます。java.util.concurrent.BlockingQueueは、要素を取得および削除する際にキューが非空になるのを待ち、要素を追加する際にキューに空きができるのを待つ操作をサポートするJavaのキューです。

JavaのBlockingQueue

JavaのBlockingQueueはnullの値を受け入れず、キューにnullの値を格納しようとするとNullPointerExceptionをスローします。JavaのBlockingQueueの実装はスレッドセーフです。すべてのキュー操作は、内部的なロックや他の形式の同時実行制御を使用して、アトミックです。JavaのBlockingQueueインターフェースは、プロデューサーとコンシューマーの問題を実装するために主に使用されます。BlockingQueueでは、プロデューサーが利用可能なスペースを待つ必要やオブジェクトがコンシューマーに利用可能になるのを待つ必要はありません。なぜなら、BlockingQueueの実装クラスがそれを処理してくれるからです。Javaは、ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、SynchronousQueueなど、いくつかのBlockingQueueの実装を提供しています。BlockingQueueでプロデューサーとコンシューマーの問題を実装する際には、ArrayBlockingQueueの実装を使用します。以下にいくつか重要なメソッドを示します。

• put(E e): このメソッドは、要素をキューに挿入するために使用されます。キューが一杯の場合、利用可能なスペースが利用可能になるまで待機します。
• E take(): This method retrieves and remove the element from the head of the queue. If queue is empty it waits for the element to be available.

今、JavaのBlockingQueueを使用して生産者と消費者の問題を実装しましょう。

JavaのBlockingQueueの例 – メッセージ

キューによって生産者によって生成され、追加される通常のJavaオブジェクトです。これをペイロードまたはキューメッセージとも呼ぶことができます。

package com.journaldev.concurrency;

public class Message {
    private String msg;
    
    public Message(String str){
        this.msg=str;
    }

    public String getMsg() {
        return msg;
    }

}

JavaのBlockingQueueの例 – 生産者

メッセージを作成し、キューに入れるProducerクラスです。

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

    private BlockingQueue queue;
    
    public Producer(BlockingQueue q){
        this.queue=q;
    }
    @Override
    public void run() {
        //メッセージを生成する
        for(int i=0; i<100; i++){
            Message msg = new Message(""+i);
            try {
                Thread.sleep(i);
                queue.put(msg);
                System.out.println("Produced "+msg.getMsg());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //終了メッセージを追加する
        Message msg = new Message("exit");
        try {
            queue.put(msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

JavaのBlockingQueueの例 – 消費者

キューからのメッセージを処理し、終了メッセージが受信されると終了するConsumerクラスです。

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable{

private BlockingQueue queue;
    
    public Consumer(BlockingQueue q){
        this.queue=q;
    }

    @Override
    public void run() {
        try{
            Message msg;
            //終了メッセージが受信されるまでメッセージを消費する
            while((msg = queue.take()).getMsg() !="exit"){
            Thread.sleep(10);
            System.out.println("Consumed "+msg.getMsg());
            }
        }catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Java BlockingQueueの例 – サービス

最後に、プロデューサーとコンシューマーのためのBlockingQueueサービスを作成する必要があります。このプロデューサーコンシューマーサービスは、固定サイズのBlockingQueueを作成し、プロデューサーとコンシューマーと共有します。このサービスはプロデューサーとコンシューマースレッドを開始し、終了します。

package com.journaldev.concurrency;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ProducerConsumerService {

    public static void main(String[] args) {
        //サイズ10のBlockingQueueを作成する
        BlockingQueue queue = new ArrayBlockingQueue<>(10);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);
        //プロデューサーを開始してキューにメッセージを生成する
        new Thread(producer).start();
        //コンシューマーを開始してキューからメッセージを消費する
        new Thread(consumer).start();
        System.out.println("Producer and Consumer has been started");
    }

}

上記のJava BlockingQueueの例プログラムの出力は以下の通りです。

Producer and Consumer has been started
Produced 0
Produced 1
Produced 2
Produced 3
Produced 4
Consumed 0
Produced 5
Consumed 1
Produced 6
Produced 7
Consumed 2
Produced 8
...

Javaのスレッドスリープは、プロデューサーとコンシューマーでメッセージを生成および消費する際に遅延を持たせるために使用されます。