当崩溃的核心客户端重新连接时，重新传递确认的消息

Question

我已经设置了一个在本地运行的独立 HornetQ 实例。出于测试目的，我使用 HornetQ core API 创建了一个消费者，它将每 500 毫秒接收一条消息。

当我的客户端连接并读取队列中的所有消息时，我在消费者端面临一个奇怪的行为，如果我强制关闭它（没有正确关闭会话/连接），那么下次我再次启动这个消费者时，它将读取队列中的旧消息。这是我的消费者示例：

// HornetQ 消费者代码

   public void readMessage() {
    ClientSession session = null;
    try {
        if (sf != null) {
            session = sf.createSession(true, true);

            ClientConsumer messageConsumer = session.createConsumer(JMS_QUEUE_NAME);
            session.start();

            while (true) {
                ClientMessage messageReceived = messageConsumer.receive(1000);
                if (messageReceived != null && messageReceived.getStringProperty(MESSAGE_PROPERTY_NAME) != null) {
                    System.out.println("Received JMS TextMessage:" + messageReceived.getStringProperty(MESSAGE_PROPERTY_NAME));
                    messageReceived.acknowledge();
                }

                Thread.sleep(500);
            }
        }
    } catch (Exception e) {
        LOGGER.error("Error while adding message by producer.", e);
    } finally {
        try {
            session.close();
        } catch (HornetQException e) {
            LOGGER.error("Error while closing producer session,", e);
        }
    }
}

有人能告诉我为什么它会这样工作吗？我应该在客户端/服务器端使用什么样的配置，这样如果消费者读取了一条消息，它就会从队列中删除它？

Answer 1

确认完成后，您没有提交会话，并且您没有创建启用了自动提交确认的会话。因此，您应该执行以下操作之一：

• 在一次或多次调用acknowledge() 后显式调用session.commit()
• 或通过使用sf.createSession(true,true) 或sf.createSession(false,true) 创建会话来启用隐式自动提交确认（控制自动提交确认的布尔值是第二个）。

请记住，当您为确认启用自动提交时，有一个内部缓冲区需要在确认刷新到代理之前达到特定大小。像这样的批处理确认可以显着提高某些大容量用例的性能。默认情况下，您需要确认 1,048,576 字节的消息才能刷新缓冲区并将确认发送到代理。您可以通过在您的ServerLocator 实例上调用setAckBatchSize 或使用不同的createSession 方法（例如sf.createSession(true, true, myAckBatchSize)）来更改此缓冲区的大小。

如果确认缓冲区未刷新并且您的客户端崩溃，则当客户端返回时，相应的消息仍将在队列中。如果缓冲区没有达到其阈值，当消费者正常关闭时，它仍然会被刷新。