Привет, Хабр!
Сегодня поговорим о паттерне Poison Pill и его реализацию в Java.
Poison Pill, или «ядовитая пилюля», — это шаблон, используемый для мягкой остановки потока или процесса. Суть его заключается в отправке специального, заранее определенного сообщения, которое сигнализирует о необходимости завершения работы. Это сообщение обрабатывается в рамках нормального потока сообщений и позволяет потоку корректно завершить работу, освободить ресурсы и закрыть все активные соединения.
Принцип работы
Архитектура паттерна Poison Pill включает несколько компонентов: Message, Producer, Consumer и MessageQueue.
Message который определяет структуру сообщений. Сообщения могут включать различные заголовки и тело сообщения. Пример реализации — SimpleMessage:
MessageQueue объединяет точки публикации MqPublishPoint и подписки MqSubscribePoint. Он представляет собой очередь, через которую сообщения передаются от производителей к потребителям. Пример реализации — SimpleMessageQueue, использующий BlockingQueue для хранения сообщений.
Producer создает сообщения и помещает их в очередь. Т. е. когда производитель завершает свою работу, он отправляет сообщение Poison Pill, чтобы уведомить потребителей о необходимости остановки.
Consumer является потребителем и извлекает сообщения из очереди и обрабатывает их. Если потребитель получает Poison Pill, он завершает свою работу
В целом, принцип работы Poison Pill основан на циклическом взаимодействии между производителями и потребителями через очередь сообщений. Производители генерируют сообщения и помещают их в очередь. Каждое сообщение обрабатывается потребителями, которые извлекают его из очереди. Когда производитель завершает работу и отправляет Poison Pill, потребитель, получив этот сигнал, прекращает обработку и завершает выполнение.
Реализация в Java
Определим интерфейс Message и класс SimpleMessage:
public interface Message {
void addHeader(String key, String value);
String getHeader(String key);
String getBody();
void setBody(String body);
}
public class SimpleMessage implements Message {
private Map<String, String> headers = new HashMap<>();
private String body;
@Override
public void addHeader(String key, String value) {
headers.put(key, value);
}
@Override
public String getHeader(String key) {
return headers.get(key);
}
@Override
public void setBody(String body) {
this.body = body;
}
@Override
public String getBody() {
return body;
}
}Определим MessageQueue и его реализации SimpleMessageQueue:
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public interface MessageQueue extends MqPublishPoint, MqSubscribePoint {}
public class SimpleMessageQueue implements MessageQueue {
private BlockingQueue<Message> queue;
public SimpleMessageQueue(int capacity) {
this.queue = new LinkedBlockingQueue<>(capacity);
}
@Override
public void put(Message message) throws InterruptedException {
queue.put(message);
}
@Override
public Message take() throws InterruptedException {
return queue.take();
}
}Реализация Producer и Consumer:
public class Producer implements Runnable {
private final MessageQueue queue;
private volatile boolean isStopped;
public Producer(MessageQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (!isStopped) {
SimpleMessage message = new SimpleMessage();
message.setBody("Important data");
queue.put(message);
Thread.sleep(1000); // Имитация работы
}
queue.put(new PoisonPillMessage()); // Отправка Poison Pill
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void stop() {
isStopped = true;
}
}
public class Consumer implements Runnable {
private final MessageQueue queue;
public Consumer(MessageQueue queue) {
this.queue = queue;
}
@Override
public void run() {
try {
Message message;
while (true) {
message = queue.take();
if (message instanceof PoisonPillMessage) break; // Остановка если Poison Pill
processMessage(message);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void processMessage(Message message) {
System.out.println("Processed: " + message.getBody());
}
}
public class PoisonPillMessage extends SimpleMessage {
public PoisonPillMessage() {
setBody("POISON_PILL");
}
}Запуск и координация потоков:
public class Main {
public static void main(String[] args) {
MessageQueue queue = new SimpleMessageQueue(10);
Producer producer = new Producer(queue);
Consumer consumer = new Consumer(queue);
Thread producerThread = new Thread(producer);
Thread consumerThread = new Thread(consumer);
producerThread.start();
consumerThread.start();
try {
Thread.sleep(5000); // позволяем производить сообщения
producer.stop(); // останавливаем производителя
producerThread.join();
consumerThread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}Альтернативы
Manual Offset Management — один из подходов к управлению отказами заключается в ручном смещении офсета до первого сообщения после Poison Pill. Так можно получить контроль над тем, какие сообщения будут пропущены, но требует доступа к Kafka или другому брокеру сообщений.
Error‑Handling Deserializers: в случае использования Apache Kafka можно применить специальный десериализатор с обработчиком ошибок. Он пропускает некорректные сообщения, тем самым избегая зацикливания при обработке данных, и передаёт сообщения об ошибках в спец. лог.
С конфигурацией Spring Kafka можно использовать SeekToCurrentErrorHandler для управления ошибками при десериализации сообщений. Обработчик смещает офсет в точку после неудачного сообщения, избегая блокировки обработки последующих сообщений.
Используя паттерн команды, можно создать единый интерфейс для различных типов задач и использовать общий механизм завершения их выполнения, что также можно адаптировать под конкретные сценарии.
А можно воспользоваться просто прямым завершением потоков. Это можно реализовать через управление потоком через класс Thread:
public class Main {
public static void main(String[] args) {
Thread worker1 = new Thread(new Worker());
Thread worker2 = new Thread(new Worker());
worker1.start();
worker2.start();
try {
Thread.sleep(1000); // даем время потокам поработать
} catch (InterruptedException e) {
e.printStackTrace();
}
// прямое завершение потоков
worker1.interrupt();
worker2.interrupt();
System.out.println("Потоки были прямо завершены.");
}
}
class Worker implements Runnable {
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
try {
System.out.println("Работа потока " + Thread.currentThread().getName());
Thread.sleep(100);
} catch (InterruptedException e) {
System.out.println("Поток " + Thread.currentThread().getName() + " прерван.");
Thread.currentThread().interrupt(); // рекомендуется восстановить статус прерванного состояния
}
}
}
}
Поток выполняет свою работу в цикле до тех пор, пока не будет прерван. Метод interrupt() используется для отправки запроса на прерывание потоку, который проверяет свое состояние с помощью Thread.currentThread().isInterrupted() .
В заключение напомню про открытый урок, который пройдет сегодня вечером в OTUS: «Переопределение, скрытие и передекларация в Java». Записаться можно по ссылке.
