アクティブMQ:古いメッセージをデキューするにはどうすればよいですか?
質問
ActiveMQ の使用方法を学んでいますが、現在次の問題に直面しています。
ActiveMQ 上に topic.test という名前のトピックがあり、2 人のサブスクライバーがいるとします。ある時点で、メッセージを待っているサブスクライバーは 1 人だけで、プロデューサーが上記のトピックに対してメッセージを送信します。
接続されているサブスクライバはメッセージを受け取りますが、他のサブスクライバは後で接続されたときにそのメッセージを受け取るべきではないでしょうか?まあ、私の場合、それは起こっていません:私の購読者は接続中にのみメッセージを受信します。接続されていない間に送信された他のメッセージはすべて受信されません。私の何が間違っているのでしょうか?
以下は、ActiveMQ をテストするために作成したソース コードの一部です。もしかしたら、何が問題なのかが見つかるかも知れません。
私の夏のコード:
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.setClientID("leitorTeste");
conexao.start();
Session sessao = conexao.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic fonte = sessao.createTopic("topic.test");
MessageConsumer consumer = sessao.createConsumer(fonte);
javax.jms.Message presente = null;
while ((presente = consumer.receive()) != null) {
System.out.println(((TextMessage) presente).getText());
}
consumer.setMessageListener(new LeitorMensagens());
conexao.close();
そして、これが私のプロデューサーコードです:
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
Session sessao = conexao.createSession(true, Session.AUTO_ACKNOWLEDGE);
connection.start();
Destination destino = sessao.createTopic("topic.test");
MessageProducer produtorMensagem = sessao.createProducer(destino);
produtorMensagem.setDeliveryMode(DeliveryMode.PERSISTENT);
TextMessage message = sessao.createTextMessage("Hi!");
produtorMensagem.send(message);
sessao.commit();
connection.close();
コンシューマーが古いメッセージを取得できるようにするために、ActiveMQ に追加する必要がある他の構成はありますか?
解決
消費者を「永続的」にする必要があります。それ以外の場合、AMQ はサブスクリプションを解除するとすぐにそのことを「忘れ」ます。これを行うには、次を使用します Session.createDurableSubscriber()
他のヒント
と呼ばれるものがあります。 遡及的な消費者 ポリシーはブローカーに設定することもできます。これは、トピック購読者向けです - 永続的ではありませんが、見逃した可能性のある「最近」のメッセージを受信したい場合があります - も参照してください。 サブスクリプション回復ポリシー