Java for Beginner

SynchronousQueue

SynchronousQueue — это специализированная реализация интерфейса BlockingQueue, который позволяет передавать элементы между потоками напрямую. В отличие от других очередей, у SynchronousQueue нет внутреннего буфера или емкости. Каждый put-запрос должен соответствовать take-запросу, чтобы операция завершилась. Это делает SynchronousQueue полезной в ситуациях, когда требуется прямая передача данных между потоками.

Внутреннее устройство SynchronousQueue

SynchronousQueue можно представить как точку передачи данных, где один поток передает элемент, и другой поток немедленно его забирает. Если нет готового потребителя для передачи, операция будет заблокирована до тех пор, пока потребитель не появится.

Блокировка и синхронизация:

SynchronousQueue использует два режима передачи данных: "fair" (справедливый) и "non-fair" (несправедливый). В справедливом режиме потоки обслуживаются в порядке поступления (FIFO), тогда как в несправедливом режиме порядок не гарантируется.
Реализация на основе блокировки и синхронизации происходит с помощью ReentrantLock и Condition.

Режимы работы:

Fair Mode: Гарантирует обслуживание потоков в порядке поступления. Это достигается за счет использования очередей ожидания для операций put и take.
Non-Fair Mode: Потоки обслуживаются без строгого порядка, что может приводить к уменьшению времени ожидания для некоторых потоков, но не гарантирует равномерности.

Основные структуры данных:

В справедливом режиме используются две очереди ожидания для хранения ожидающих put и take операций.
В несправедливом режиме используется стек для выполнения операций.

Пример кода внутреннего устройства

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

public class SynchronousQueueExample {
    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>(true); // fair mode

        Thread producer = new Thread(() -> {
            try {
                System.out.println("Putting element: 1");
                queue.put(1);
                System.out.println("Element 1 put");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                System.out.println("Taking element...");
                Integer element = queue.take();
                System.out.println("Element taken: " + element);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

Особенности SynchronousQueue

Отсутствие буферизации: У SynchronousQueue нет внутреннего буфера, что означает, что каждое добавление элемента должно совпадать с извлечением.
Потокобезопасность: Очередь разработана с учетом многопоточности и обеспечивает высокую степень безопасности для работы в конкурентной среде.
Высокая скорость передачи данных: За счет отсутствия буферизации и немедленной передачи элементов достигается высокая скорость обмена данными между потоками.
Гибкость: Возможность выбора между справедливым и несправедливым режимами работы позволяет оптимизировать производительность в зависимости от конкретных требований.

Преимущества SynchronousQueue

Эффективность в межпоточном взаимодействии: SynchronousQueue идеально подходит для ситуаций, где требуется немедленная передача данных между потоками, например, для реализации паттерна "производитель-потребитель".
Простота реализации: Очередь предоставляет простой API для обмена данными между потоками, что упрощает разработку многопоточных приложений.
Высокая производительность: За счет минимального количества операций по синхронизации достигается высокая производительность при обмене данными.

Ссылки на полезные статьи (спасибо авторам за проделанную работу) :
https://www.baeldung.com/java-synchronous-queue

#Java #Training #Medium #SynchronousQueue

Baeldung

A Guide to Java SynchronousQueue | Baeldung

Learn how you can utilize Java's SynchronousQueue to help you coordinate the work of multiple threads.

938 viewsДен, 05:32

Java for Beginner

Основные методы SynchronousQueue и примеры использования

put(E e):

Добавляет элемент в очередь. Если нет готового потребителя, операция блокируется до тех пор, пока потребитель не появится.

SynchronousQueue<Integer> queue = new SynchronousQueue<>();
queue.put(1); // блокируется до появления потребителя

take():

Извлекает и удаляет элемент из очереди. Если нет готового поставщика, операция блокируется до тех пор, пока элемент не будет добавлен.

Integer element = queue.take(); // блокируется до появления элемента

offer(E e, long timeout, TimeUnit unit):

Пытается добавить элемент в очередь в течение заданного времени ожидания. Возвращает true, если элемент был добавлен, и false в противном случае.

boolean success = queue.offer(1, 2, TimeUnit.SECONDS); // попытка добавить элемент в течение 2 секунд

poll(long timeout, TimeUnit unit):

Пытается извлечь элемент из очереди в течение заданного времени ожидания. Возвращает элемент или null, если время ожидания истекло.

Integer element = queue.poll(2, TimeUnit.SECONDS); // попытка извлечь элемент в течение 2 секунд

isEmpty():

Возвращает true, если очередь пуста.

boolean empty = queue.isEmpty();

size():

Возвращает количество элементов в очереди. В случае SynchronousQueue всегда возвращает 0 или 1.

int size = queue.size();

Примеры использования SynchronousQueue

Производитель-потребитель:
SynchronousQueue часто используется для реализации паттерна "производитель-потребитель", где производитель и потребитель обмениваются данными напрямую.

import java.util.concurrent.SynchronousQueue;

public class ProducerConsumerExample {
    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    System.out.println("Produced: " + i);
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    Integer item = queue.take();
                    System.out.println("Consumed: " + item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

Синхронизация потоков:
SynchronousQueue можно использовать для синхронизации потоков, обеспечивая передачу сигналов между ними.

import java.util.concurrent.SynchronousQueue;

public class ThreadSyncExample {
    public static void main(String[] args) {
        SynchronousQueue<String> queue = new SynchronousQueue<>();

        Thread thread1 = new Thread(() -> {
            try {
                System.out.println("Thread 1 waiting for signal...");
                String signal = queue.take();
                System.out.println("Thread 1 received signal: " + signal);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread thread2 = new Thread(() -> {
            try {
                Thread.sleep(2000);
                queue.put("Start");
                System.out.println("Thread 2 sent signal");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        thread1.start();
        thread2.start();
    }
}

#Java #Training #Medium #SynchronousQueue

313 viewsДен, 14:34

About

Blog

Apps

Platform