从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接口而不是使用receive方法，或者您可以从receive方法获取消息并异步地在线程池中处理它们，您不希望阻止收到消息。

当您创建消费者可以添加监听器，如下：

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()); 
    } 
}