消費者からプロデューサに承認を送って、それをactivemqとrabbitmqで処理します。

Question

私が知っているように、ActiveMQには実際にメッセージが受信されたことをプロデューサに知らせるAUTO Acknowledgeと呼ばれる機能があります。

ActiveMQまたはRabbitMQのコンシューマからプロデューサに肯定応答を送信できるかどうかを知りたい。プロデューサで確認応答メッセージを処理し、確認応答を受信しない場合は、メッセージを再度コンシューマに送信します。

Answer 1

非同期メディアで同期ユースケースを実行します。ここで説明したように、RPCを使用することができるのRabbitMQの場合

からhttps://www.rabbitmq.com/tutorials/tutorial-six-python.html と https://www.rabbitmq.com/direct-reply-to.html

も、著者はそれを避けるために助言することに注意してください：疑いでRPCを避けるとき

を。可能であれば、RPCのようなブロッキングではなく、非同期パイプラインを使用する必要があります。結果は非同期的に次の計算ステージにプッシュされます。

---

RabbitMQのJavaクライアントはcom.rabbitmq.client.Channel.basicConsumeを通じて自動肯定応答（ACK）を提供します。

Answer 2

少なくともActiveMQのために - 。これが内蔵されてあなたは、単にあなたが消費されるメッセージを監視したいキューの助言のトピックを聞くactivemq.xmlに

<policyEntry queue=">" advisoryForConsumed="true"/> 

それをオンにする必要があります。次に、メッセージIDを抽出することができます。また、未処理の要求をチェックすることもできます。

完全なエンドツーエンドの承認については、私はもっとカスタムを推奨します。私。 producer-appは、生成されたメッセージのステータスに関する応答を受け取る "応答"キューを聴かなければなりません。私。処理が失敗した場合など理由を知りたい場合があります。

とにかく、プロデューサーがActiveMQからの肯定応答を聞くコードもあります。

public void run() throws Exception { 
    ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616"); 
    conn = cf.createConnection(); 
    sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); 
    Destination dest = sess.createQueue("duck"); 
    MessageConsumer mc = sess.createConsumer(AdvisorySupport.getMessageConsumedAdvisoryTopic(dest)); 
    mc.setMessageListener(this); 
    conn.start(); 

    MessageProducer mp = sess.createProducer(sess.createQueue("duck")); 
    mp.send(sess.createTextMessage("quack")); 
} 


public void onMessage(Message msg) { 
    try { 
     String msgId = msg.getStringProperty("orignalMessageId"); 
     System.out.println("Msg: " + msgId + " consumed"); 
    } catch (Exception e) { 
     e.printStackTrace(); 
    } 
}