消费来自 JMS 主题的消息时延迟

Question

我有一个长期订阅者的主题。我可以发布和使用消息，但是我发现从主题中读取消息时存在一些延迟。

我无法在一次通话中阅读这些消息。我需要多次调用该方法来阅读消息。我错过了什么吗？

    private void publishMessage() {
        TopicConnection topicConnection = null;
        TopicSession topicSession = null;
        TopicPublisher topicPublisher = null;
        try {
          topicConnection = connectionFactory.createTopicConnection();
          topicSession = topicConnection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
          Topic topicName= topicSession.createTopic(topicName);
          topicPublisher = topicSession.createPublisher(topicName);
          ObjectMessage message = topicSession.createObjectMessage(customObject)
          message.setStringProperty("user", userProperty);
          topicPublisher.publish(message, DeliveryMode.PERSISTENT, Message.DEFAULT_PRIORITY, timeToLive);
        } catch (JMSException e) {
          throw new RuntimeException("Error Sending UMessage", e);
        } finally {
          closeConnections(null, topicPublisher, topicSession, topicConnection);
        }
    }

public void consumeMessages(String userId, int maxResults) {
    TopicConnection topicConnection = null;
    TopicSession topicSession = null;
    TopicSubscriber topicSubscriber = null;

    try {
      topicConnection = connectionFactory.createTopicConnection("guest","guest");
      topicConnection.setClientID("topic");
      topicSession = topicConnection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
      Topic topicName= topicSession.createTopic(topicName);
      topicSubscriber = topicSession.createDurableSubscriber(topicName, "subscriptionname", String.format("user = '%s'", userName), false);
      topicConnection.start();
      Message msg = null;

      do {
        msg = topicSubscriber.receiveNoWait();
        if (msg instanceof ObjectMessage) {
          ObjectMessage om = (ObjectMessage) msg;
         else {
            log.error(String.format(" %s", om.getObject().getClass().getSimpleName()));
          }
        } else if (msg != null) {
          log.error(String.format("e %s", msg.getClass().getSimpleName()));
        }
      } while (msg != null && out.size() <= maxResults);
    } catch (JMSException e) {
      throw new RuntimeException("Error retrieving User Messages", e);
    } finally {
      closeConnections(topicSubscriber, null, topicSession, topicConnection);
    }
    return out;
}

Answer 1

您正在调用 receiveNoWait()，它将一次检索一条消息。通常情况下，取决于您的 JMS 提供者，JMS 客户端将一次检索多个消息并将它们缓存在客户端以减少网络延迟。当您调用接收时，它会从此缓存中获取消息并将其提供给您。

如果您看到一分钟的延迟，可能是您将这些消息放到主题上的方式有问题，或者您在处理每条消息时阻止了您的消息接收。如果您不想在处理时阻止消息接收，请查看实现MessageListener 接口而不是使用接收方法，或者您可以从接收方法中获取消息并在线程池上异步处理它们。

当你创建消费者时，你可以像这样添加监听器：

MessageListener listener = new MyListener();
consumer.setMessageListener(listener);

然后创建一个类来处理消息或在您现有的消费者类中实现接口：

public class MyListener implements MessageListener {
  public void onMessage(Message message)
  {
     TextMessage text = (TextMessage) message;

     System.out.println("Message: " + text.getText());
  }
}