Производитель потребительская модель в JAVA

Question

Here the pattern of consumer-producer

Producer создает новое сообщение, consumer слушает их через Broker.

Пока producer создает и отправляет сообщение, а consumer ждет сообщения. Когда приходит новое сообщение, consumer выводит его на экран.

Я хочу, чтобы Producer был уведомлен о том, что Consumer получил сообщение и его освободил.

Интересно, есть ли способ уведомить/подать заявку Producer от Consumer? Или я должен сообщить Consumer через Broker? Как это можно сделать?

Answer 1

Стандарт JMS поддерживает очередь ответа (JMSReplyTo), так что исходный потребитель может вернуть ответное сообщение исходному производителю в конкретной очереди.

Как показано здесь;

http://www.enterpriseintegrationpatterns.com/patterns/messaging/RequestReplyJmsExample.html

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

Это означает, что потребитель станет производителем для ответного сообщения.

ссылается также имеют код, чтобы показать рабочий пример (хотя я не пробовал сам)

Answer 2

Я бы спросил, почему Продюсер должен знать.

Вся цель производителя/потребителя состоит в том, чтобы отделить два объекта. Производители не знают и не заботятся о том, кто обрабатывает сообщение.

Ваше предложение разбивает эту модель. Вы не должны этого делать, если у вас нет веской причины, кроме «Я хочу».

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

Но теперь эти два безнадежно запутались. Я думаю, что это плохой выбор.

Answer 3

Это больше похоже на Слушатель узора. Используйте EventBus. 2 события: public PutEvent(MyType message) {...}, public TakeEvent(long id) {...}. Тогда:

EventBus bus = new SimpleEventBus(); 
bus.addHandler(producerInstance::onMessageTaken, GetEvent.getType()); 
bus.addHandler(consumerInstance::onMessageArrived, PutEvent.getType()); 

Потребитель:

MyType message = queue.pop(); 
bus.fireEventFromSource(new TakeEvent(message.getId()), this); 
... 
@Override public void onMessageArrived(PutEvent event) { 
    // We know, that there're messages out there and can later pop it 
    System.out.printf("%d from %s", event.getMessage().getId(), event.getSource() != null ? event.getSource().hashCode() + "" : "unknown"); 
} 

Производитель:

MyType message = new MyType(data, ++id); 
queue.add(message); 
bus.fireEventFromSource(new PutEvent(message), this); 
... 
@Override public void onMessageTaken(TakeEvent event) { 
    // Wow, who took it? 
    System.out.printf("%s took %d", event.getSource() != null ? event.getSource().hashCode() + "" : "unknown", event.getId()); 
}