4

我在 hornetQ 中与一位消费者创建了会话，然后我使用生产者在队列中添加了 4 条消息。在此之后，我创建了新的消费者。

这个消费者会知道旧消息吗？

如果不是，是否可以在 XML 中配置它？

我创建了无法获取以前消息的新消费者。我只是想确认这种行为是否正确？我在文档中没有找到任何帮助。

下面是代码片段：

TextMessage receivedMessage = (TextMessage)consumer.receive(); 
receivedMessage.acknowledge();
System.out.println("Got order: " + receivedMessage.getText());
//consumer.close();
MessageConsumer newConsumer = session.createConsumer(orderQueue);
receivedMessage = (TextMessage)newConsumer.receive();
receivedMessage.acknowledge();  
System.out.println("Got order: " + receivedMessage.getText());


如果我取消注释 consumer.close() 行，它工作正常。
我的 hornetq-jms.xml

<connection-factory name="NettyConnectionFactory">
      <xa>true</xa>
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="/XAConnectionFactory"/>
      </entries>
       <consumer-window-size>0</consumer-window-size>
   </connection-factory>

   <connection-factory name="NettyConnectionFactory">
      <xa>false</xa>
      <connectors>
         <connector-ref connector-name="netty"/>
      </connectors>
      <entries>
         <entry name="/ConnectionFactory"/>
      </entries>
       <consumer-window-size>0</consumer-window-size>
   </connection-factory>

   <connection-factory name="NettyThroughputConnectionFactory">
      <xa>true</xa>
      <connectors>
         <connector-ref connector-name="netty-throughput"/>
      </connectors>
      <entries>
         <entry name="/XAThroughputConnectionFactory"/>
      </entries>
       <consumer-window-size>0</consumer-window-size>
   </connection-factory>

   <connection-factory name="NettyThroughputConnectionFactory">
      <xa>false</xa>
      <connectors>
         <connector-ref connector-name="netty-throughput"/>
      </connectors>
      <entries>
         <entry name="/ThroughputConnectionFactory"/>
      </entries>
       <consumer-window-size>0</consumer-window-size>
   </connection-factory>

连接工厂代码片段

TransportConfiguration transportConfiguration = new
                TransportConfiguration(NettyConnectorFactory.class.getName());
        HornetQConnectionFactory cf =
                HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF,getTransportConfiguration());
        Queue orderQueue = HornetQJMSClient.createQueue("MutationPipelineQueue");

getTransportConfiguration() 的代码：

private synchronized static TransportConfiguration getTransportConfiguration() {
        HashMap<String, TransportConfiguration> transportConfigurationMap = new HashMap<String, TransportConfiguration>();
        TransportConfiguration tc = transportConfigurationMap.get("machinename:5455");
        if(tc == null){
            Map<String, Object> connectionParams = new HashMap<String, Object>();
            connectionParams.put(org.hornetq.core.remoting.impl.netty.TransportConstants.HOST_PROP_NAME,"machinename");
            connectionParams.put(org.hornetq.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME,Integer.valueOf("5455"));
            tc = new TransportConfiguration(NettyConnectorFactory.class.getName(), connectionParams);
            transportConfigurationMap.put("machinename:5455", tc);
        }
        return tc;
4

2 回答 2

7

是的，它会知道您的旧消息。但是在这种情况下，您的旧消费者仍然打开，因此消费者将在其缓冲区中缓存消息，除非您关闭它，或者您更改消费者窗口大小 = 0。

大多数消息系统将在消费者上提前缓存，因此下次您在消费者上调用接收时，消息将准备好被接收。

但是，如果您的消费者速度很慢并且您没有那么多消息，则该消息将在客户端的缓冲区中，直到您关闭该消费者。

对于生产中的快速消费者来说，最好始终提前缓存，因为这将提高您的吞吐量，而这将受到网络延迟的限制而没有缓存。

在 HornetQ 案例中，您可以通过设置 consumer-window-size=0 来应对缓慢的消费者。

http://docs.jboss.org/hornetq/2.3.0.beta3/docs/user-manual/html/flow-control.html#flow-control.consumer.window

如果您通过 JNDI 查找实例化连接工厂：

   <connection-factory name="ConnectionFactory">
      <connectors>
         <connector-ref connector-name="netty-connector"/>
      </connectors>
      <entries>
         <entry name="ConnectionFactory"/>
      </entries>

      <!-- We set the consumer window size to 0, which means messages are not buffered at all
      on the client side -->
      <consumer-window-size>0</consumer-window-size>

   </connection-factory>

或者在您直接实例化连接工厂的情况下，您必须在实例中设置 consumerWindowSize：

TransportConfiguration transportConfiguration = new
                TransportConfiguration(NettyConnectorFactory.class.getName());
HornetQConnectionFactory cf =
HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF,getTransportConfiguration());
cf.setConsumerWindowSize(0) // <<<<<< here

这是一个来自 HornetQ 发行版的运行示例，位于 examples/jms/no-consumer-buffering。它与您的代码片段完全相同，并且每次都有效：

 // Step 5. Create a JMS Session
 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

 // Step 6. Create a JMS Message Producer
 MessageProducer producer = session.createProducer(queue);

 // Step 7. Create a JMS MessageConsumer

 MessageConsumer consumer1 = session.createConsumer(queue);

 // Step 8. Start the connection

 connection.start();

 // Step 9. Send 10 messages to the queue

 final int numMessages = 10;

 for (int i = 0; i < numMessages; i++)
 {
    TextMessage message = session.createTextMessage("This is text message: " + i);

    producer.send(message);
 }


 System.out.println("Sent messages");

 // Step 10. Create another consumer on the same queue

 MessageConsumer consumer2 = session.createConsumer(queue);

 // Step 11. Consume three messages from consumer2

 for (int i = 0; i < 3; i++)
 {
    TextMessage message = (TextMessage)consumer2.receive(2000);

    System.out.println("Consumed message from consumer2: " + message.getText());
 }

正如您在此示例中看到的那样，正在接收旧消息。

任何与此不同的东西都是您系统的错误配置。也许您没有设置正确的连接工厂？

顺便说一句：在 ActiveMQ 上，您可以管理预取限制以管理相同的行为：

http://activemq.apache.org/what-is-the-prefetch-limit-for.html

这个问题是具有多个消费者的 JMS 队列的精确复制

至于追溯消息，这是 ActiveMQ 上仅适用于主题的另一个功能，它是使用旧消息创建的订阅。

于 2013-03-04T15:16:46.223 回答
3

您正在寻找的功能是通过 Durable 订阅提供的。这是标准 JMS 规范的一部分。如果您查看您的版本的 HornetQ 文档，我相信您会找到它。此外，这里是在 HornetQ 中使用 JMS Durable 订阅的 Java 客户端的一个很好的示例。

于 2013-03-04T12:56:26.323 回答