Пример Java BlockingQueue

Сегодня мы рассмотрим Java BlockingQueue. java.util.concurrent.BlockingQueue – это очередь в Java, которая поддерживает операции, ожидающие, когда очередь становится непустой при извлечении и удалении элемента, а также ожидающие, когда в очереди появляется свободное место при добавлении элемента.

Java BlockingQueue

Java BlockingQueue не принимает значения null и генерирует исключение NullPointerException, если вы пытаетесь сохранить значение null в очереди. Реализации Java BlockingQueue являются потокобезопасными. Все методы внутри очереди атомарны по своей природе и используют внутренние блокировки или другие формы контроля конкурентности. Интерфейс Java BlockingQueue является частью коллекций Java и в основном используется для решения проблемы производителя и потребителя. Нам не нужно беспокоиться ожиданием места для производителя или объекта для потребителя в BlockingQueue, потому что об этом заботятся реализации классов BlockingQueue. В Java предоставляется несколько реализаций BlockingQueue, таких как ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue и др. При реализации проблемы производителя и потребителя в 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 – Производитель

Класс производителя, который будет создавать сообщения и помещать их в очередь.

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 – Потребитель

Класс потребителя, который будет обрабатывать сообщения из очереди и завершаться, когда будет получено сообщение об окончании.

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();
        }
    }
}

Пример BlockingQueue на Java – Сервис

Наконец, нам нужно создать сервис 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) {
        //Создание BlockingQueue размером 10
        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");
    }

}

Вывод приведенной выше программы примера BlockingQueue на Java показан ниже.

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 Thread sleep используется в производителе и потребителе для размещения и потребления сообщений с некоторой задержкой.