java - Nack 当前消息之前的所有消息并 Ack 当前消息(rabbitmq，java)

Question

channel.basicQos(1);

while (true) {
    GetResponse res = channel.basicGet(TEST_QUEUE, false);
    if (res != null) {
        deliveryTag = res.getEnvelope().getDeliveryTag();
    }

    // Handle all messages If the condition is true
    if (condition) {
        // nack all messages unhandled previously
        channel.basicNack(deliveryTag - 1, true, true);

        // ack current message only
        channel.basicAck(deliveryTag, false);
    }
    else {
        // Do not handle current message and continue to get next one
    }           
}   

Q1。
我不确定是否可以同时使用 nack 和 ack。
我可以使用 DeliveryTag - 1 来指示所有以前的消息吗？

简而言之，我想跳过所有不满足if条件的消息。
如果当前消息满足条件，则 nack 所有跳过的消息并确认当前消息。
通过这样做，我想延迟处理一些特定的消息。

第二季度。
恐怕如果我写为 while (true) 并且有多个工作人员正在运行，那么 channel.basicQos(1) 将无法按预期工作。
我应该编写这样的代码来限制计数吗？或者我应该如何编写才能确保所有其他工作人员都能均匀地获取消息？

int prefetch = 1;
int count = 0;
while (count++ <= prefetch) {
}

第三季度。
我注意到只要连接打开，工作程序就不会终止。
连接将打开多长时间？我是否需要手动关闭它？

最后，RabbitMQ java client API vs AmqpTemplate vs RabbitTemplate 哪一个更适合这种情况(不使用 MessageListener(ChannelAwareMessageListener) 模型)？

Answer 1

Q1 - 它应该可以正常工作。您尝试过并发现问题吗？是的，每次交付时标签都会递增。

Q2 - basicQos 与 basicGet() 无关 - 它仅与 basicConsume() 一起使用。

Q3 - 完成后您需要关闭连接。

最后；这取决于。如果你想要 Spring 的更高级别的支持(消息转换等)，那么就使用它；如果您想处理原始数据API，请不要使用Spring。

RabbitTemplate 不直接支持具有用户管理的 ack/nack 的 basicGet，除非通过其带有 channel 回调的 execute 方法。