java - 如何暂停和恢复 JMS 消息的异步消费

标签 java jms activemq jms-topic

我正在使用 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/

相关文章:

java - Android 位图动画

Java 静态导入导致编译错误。可能的编译器错误?

java - 如何在 Java 中将 JMSBytesMessage 转换为 TextMessage

encryption - 使用 JMS 请求-答复的密码安全

java - 如何创建 REST API,然后 REST API 将对象/字符串放入 ActiveMQ

glassfish - 使用故障转移协议(protocol)时使用 genericra 集成 activemq 和 glassfish 时出现问题

java - 终结符和非终结符

java - 匹配以(正则表达式)java结尾的字符串

java - 与 Play 框架一起使用的 JMS/ActiveMQ 异常

multithreading - 并发消息处理设计争议