3

对此可能有一个愚蠢的简单答案，但我正在尝试使用 ActiveMQ 在生产者和消费者之间传递消息。我将有许多生产者和许多消费者，但我希望每条消息在消费者之间只传递一次。这似乎意味着我不能使用主题，因为它们会将消息传递给所有正在收听的消费者，而我希望只有一个消费者接收每条消息。

我的问题是我能够接收消息，但消息没有出队。所以如果我重新启动我的消费者进程，所有的消息都会被重新处理。 这个答案似乎是相关的，但似乎并不适用，因为我无法创建持久的队列消费者，只能创建持久的主题消费者（除非我在 API 文档中遗漏了一些东西）。

我的代码如下。

TopicConnectionFactory factory = new ActiveMQConnectionFactory(props.getProperty("mq.url"));
Connection conn  = factory.createConnection();
Session session = conn.createSession(true, Session.CLIENT_ACKNOWLEDGE);
Queue queue = session.createQueue(props.getProperty("mq.source_queue"));
conn.start();
MessageConsumer consumer = session.createConsumer(queue);

然后稍后

Message msg = consumer.receive();
msg.acknowledge();
if (!(msg instanceof TextMessage)) continue;
String msgStr = ((TextMessage)msg).getText();

这是我当前的代码。我尝试过使用 Session.AUTO_ACKNOWLEDGE 而没有使用 msg.acknowledge()。此代码的任何工作排列似乎都可以检索消息，但是当我重新启动消费者时，所有消息都会再次收到，即使它们是在重新启动之前收到的。

4

1 回答 1

5

您将会话创建为事务会话，因此如果您想通知代理所有消息现在都已使用并且不需要重新传递，则需要调用 session.commit。如果您没有将 createSession 的第一个参数设置为 true，则尊重 Ack 模式，否则将忽略它，这是 JMS API 的一个奇怪之处。如果你这样做：

ConnectionFactory factory = new ActiveMQConnectionFactory(props.getProperty("mq.url"));
Connection conn  = factory.createConnection();
Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
Queue queue = session.createQueue(props.getProperty("mq.source_queue"));
conn.start();
MessageConsumer consumer = session.createConsumer(queue);

那么这将起作用：

Message msg = consumer.receive();
msg.acknowledge();

否则你需要这样做：

Message msg = consumer.receive();
session.commit();

但是请记住，对于单个消息事务来说，客户端确认没有事务是一个更好的选择。

于 2012-06-30T20:17:24.860 回答