我正在使用 activeMQ 构建一个应用程序,其中有一个生产者和一个消费者。 在消费者中,我使用 MessageListener 异步监听来自生产者的消息,这是使用名为 onMessage(Message message) 的方法完成的。 但在使用消息之前,我想执行条件检查,然后使用消息。 我不想使用消息的同步消费,因为这违背了我的设计。
void initialize() throws JMSException {
this.connection = this.connectionFactory.createConnection();
this.connection.start();
final Session session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
final Destination destination = session.createQueue("testQ");
this.consumer = session.createConsumer(destination);
this.consumer.setMessageListener(this);
}
检查此处的状况,例如检测互联网连接等
public void onMessage(final Message message) {
Preconditions.checkNotNull(message);
if (!(message instanceof TextMessage)) {
_LOG.error("The message is not of type TextMessage but of type {} so we could not process", message.getClass().getSimpleName());
throw new IllegalArgumentException("This type '" + message.getClass().getSimpleName() + "' of message could not been handled");
}
try {
final String messageType = message.getStringProperty("messageType");
Preconditions.checkNotNull(messageType);
_LOG.info("The MessageType is {}", messageType);
final String msg = ((TextMessage) message).getText();
Preconditions.checkNotNull(msg);
_LOG.debug(msg);
process(messageType, msg);
} catch (final JMSException e) {
_LOG.error("We could not read the message", e);
}
}
任何代码示例都很棒。
最佳答案
目前尚不清楚为什么要进行检查,因为如果连接出现任何问题,那么您的应用程序将收到此类错误的通知。您可以为 JMS 连接设置一个 ExceptionListener,如果连接出现问题,它将被调用。
如果存在连接问题,则不会调用 onMessage。
此外,我会在为消费者设置消息监听器后推送 connection.start() 方法调用。 Connection.start() 调用向消息传递提供程序表明应用程序已准备好接收消息。
有connection.stop()来暂停/停止消息传递。您可以再次发出connection.start()来恢复消息传递。
<小时/>根据您的回复进行更新
我建议您使用带有客户端确认模式的 session 。
final Session session = this.connection.createSession(false, Session.CLIENT_ACKNOWLEDGE)
然后在onMessage
方法中,如果没有互联网连接,则不确认该消息。如果消息未过期,将会再次发送。
关于java - 如何暂停和恢复 JMS 消息的异步消费,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39571735/