SynchronousQueue — это особый вид BlockingQueue, в котором каждая операция вставки должна ожидать соответствующей операции удаления другим потоком, и наоборот. Когда вы вызываете метод put () для SynchronousQueue, он блокируется, пока не появится другой поток, чтобы вывести этот элемент из очереди. Точно так же, если поток пытается удалить элемент, и в данный момент элемент отсутствует, этот поток блокируется, пока другой поток не поместит элемент в очередь. Вы можете соотнести SynchronousQueue со спортсменами ( потоками ), бегущими с олимпийским факелом, они бегут с факелом (объект должен быть пропущен) и передают его другому спортсмену, ожидающему на другом конце. Если вы обратите внимание на имя, вы также поймете, что оно называется SynchronousQueue с указанием причины, оно передает данные синхронно в другой поток; он ждет, пока другая сторона получит данные, а не просто помещает данные и возвращает их (асинхронная операция). Если вы знакомы с CSP и Ada, то вы знаете, что синхронные очереди похожи на каналы рандеву. Они хорошо подходят для схем ручной передачи, в которых объект, запущенный в одном потоке, должен синхронизироваться с объектом, запущенным в другом потоке, чтобы передать ему некоторую информацию, событие или задачу. В более ранних обучающих программах по многопоточности мы узнали, как решить проблему потребителя производителя, используя ожидание и уведомление , и BlockingQueue, а в этом руководстве мы узнаем, как реализовать шаблон проектирования потребителя производителя, используя синхронную очередь . Этот класс также поддерживает необязательную политику справедливости для упорядочивания ожидающих потоков производителей и потребителей. По умолчанию этот порядок не гарантируется. Однако очередь, созданная со свойством fairness, установленным в true, предоставляет потокам доступ в порядке FIFO.
Решение Consumer для производителей с использованием SynchronousQueue в Java
Как я уже говорил, нет ничего лучше, чем проблема производителя-потребителя для понимания межпотоковой коммуникации на любом языке программирования. В проблеме потребителя источника один поток действует как производитель, который создает событие или задачу, а другой поток действует как потребитель. Общий буфер используется для передачи данных от производителя к потребителю. Сложность в решении проблемы потребителя производителя связана с крайними случаями, например, производитель должен ждать, если буфер заполнен, или поток потребителя должен ждать, если буфер пуст. Позже это было довольно просто, поскольку блокирующая очередь предоставляет не только буфер для хранения данных, но и управление потоком для блокировки потока, вызывающего метод put () (PRODUCER), если буфер заполнен, и блокирующий поток, вызывающий метод take () (CONSUMER), если буфер пуст , В этом руководстве мы решим ту же проблему с помощью SynchronousQueue, особого вида параллельной коллекции, которая имеет нулевую емкость.
В следующем примере у нас есть два потока с именами PRODUCER и CONSUMER (вы всегда должны называть свои потоки, это одна из лучших практик написания параллельных приложений). Первый поток, публикующий счет крикета, и второй поток потребляют это. Здесь результаты крикета — не что иное, как объект String. Если вы запустите программу как есть, вы не заметите ничего другого. Чтобы понять, как работает SynchronousQueue и как он решает проблему с потребителем производителя , вам нужно либо отладить эту программу в Eclipse, либо просто запустить поток производителя, комментируя consumer.start (); Если потребительский поток не запущен, тогда производитель заблокирует
очередь. ставить (событие); позвоните, и вы не увидите опубликованное событие [PRODUCER]: ЧЕТЫРЕ. Это происходит из-за особого поведения
SynchronousQueue, который гарантирует, что поток, вставляющий данные, будет блокироваться до тех пор, пока не найдется поток, удаляющий эти данные, или наоборот. Вы можете проверить другую часть кода, комментируя продюсера. Начало(); и только начинающий потребительский поток.
|
01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
import java.util.concurrent.SynchronousQueue;/** * Java Program to solve Producer Consumer problem using SynchronousQueue. A * call to put() will block until there is a corresponding thread to take() that * element. * * @author Javin Paul */public class SynchronousQueueDemo{ public static void main(String args[]) { final SynchronousQueue<String> queue = new SynchronousQueue<String>(); Thread producer = new Thread("PRODUCER") { public void run() { String event = "FOUR"; try { queue.put(event); // thread will block here System.out.printf("[%s] published event : %s %n", Thread .currentThread().getName(), event); } catch (InterruptedException e) { e.printStackTrace(); } } }; producer.start(); // starting publisher thread Thread consumer = new Thread("CONSUMER") { public void run() { try { String event = queue.take(); // thread will block here System.out.printf("[%s] consumed event : %s %n", Thread .currentThread().getName(), event); } catch (InterruptedException e) { e.printStackTrace(); } } }; consumer.start(); // starting consumer thread }}Output:[CONSUMER] consumed event : FOUR [PRODUCER] published event : FOUR |
Если вы внимательно отправили вывод, вы бы заметили, что порядок событий обратный. Кажется, поток [CONSUMER] потребляет данные еще до того, как поток [PRODUCER] его создал. Это происходит потому, что по умолчанию SynchronousQueue не гарантирует какого-либо порядка, но имеет политику честности, которая при значении true разрешает доступ к потокам в порядке FIFO. Вы можете включить эту политику справедливости, передав true перегруженному конструктору SynchronousQueue, т.е. новому SynchronousQueue (логическое справедливое).
Что нужно помнить о SynchronousQueue в Java
Вот некоторые важные свойства этой специальной очереди блокировки в Java. Очень полезно синхронно передавать данные из одного потока в другой. У него нет емкости и блоков, пока на другом конце не будет резьбы.
- SynchronousQueue блокируется, пока другой поток не будет готов принять элемент, который пытается установить один поток.
- SynchronousQueue имеет нулевую емкость.
- SynchronousQueue используется для реализации стратегии очередей прямой передачи обслуживания, при которой поток переключается на ожидающий поток, в противном случае создается новый, если это разрешено, иначе задача отклоняется.
- Эта очередь не разрешает нулевые элементы, добавление нулевых элементов приведет к исключению NullPointerException .
- Для целей других методов Collection (например, содержит) SynchronousQueue действует как пустая коллекция.
- Вы не можете заглянуть в синхронную очередь, потому что элемент присутствует только при попытке удалить его; Точно так же вы не можете вставить элемент (используя любой метод), если другой поток не пытается удалить его.
- Вы не можете выполнять итерацию по SynchronousQueue, поскольку итерировать нечего.
- SynchronousQueue, созданный с политикой справедливости, установленной в true, предоставляет потокам доступ в порядке FIFO.
Это все о SynchronousQueue в Java . Мы увидели некоторые особые свойства этой специальной параллельной коллекции и узнали, как решить классическую проблему производителя с использованием SynchronousQueue в Java. Кстати, называть ее Очередь немного запутанно, потому что она не способна удержать ваш элемент. Операция call to put () не будет завершена, пока не найдется поток, который вызывает операцию take (). Лучше быть местом встречи между потоками, чтобы делиться объектами. Другими словами, это утилита для синхронного обмена данными между двумя потоками в Java, вероятно, более безопасная альтернатива методам ожидания и уведомления .
| Ссылка: | Пример SynchronousQueue на Java — решение для потребителей от нашего партнера по JCG Явина Пола в блоге Javarevisited . |