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

Question

我正在使用 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);
        }
    }

任何代码示例都很棒。

Answer 1

目前尚不清楚为什么要进行检查，因为如果连接出现任何问题，那么您的应用程序将收到此类错误的通知。您可以为 JMS 连接设置一个 ExceptionListener，如果连接出现问题，它将被调用。

如果存在连接问题，则不会调用 onMessage。

此外，我会在为消费者设置消息监听器后推送 connection.start() 方法调用。 Connection.start() 调用向消息传递提供程序表明应用程序已准备好接收消息。

有connection.stop()来暂停/停止消息传递。您可以再次发出connection.start()来恢复消息传递。

<小时/>

根据您的回复进行更新

我建议您使用带有客户端确认模式的 session 。

final Session session = this.connection.createSession(false, Session.CLIENT_ACKNOWLEDGE)

然后在onMessage方法中，如果没有互联网连接，则不确认该消息。如果消息未过期，将会再次发送。