兔子消息确认答案

【问题标题】：rabbit messaging confirmation兔子消息确认
【发布时间】：2014-06-30 15:00:34
【问题描述】：

我正在使用rabbitmq，我想确保如果我在客户端出现连接问题，我发布的消息不会丢失。我用 eclipse 模拟它：我在 100 条消息后执行 system.exit 获取程序。我发布了 1000 条消息。第二次运行我不限制消息的数量，它返回了 3 次 840 条消息。你能帮助我吗？

生产者的代码是：

public void run() {
    String json =SimpleQueueServiceSample.getFromList();
    while (!(json.equals(""))){
        json =SimpleQueueServiceSample.getFromList();
        try {   
            c.basicPublish("", "test", 
                    MessageProperties.PERSISTENT_TEXT_PLAIN, json.getBytes());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    try {
        c.waitForConfirmsOrDie();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

消费者的代码是：

 QueueingConsumer consumer = new QueueingConsumer(channel);
 channel.basicConsume(QUEUE_NAME, true, consumer);  
  while (true) {
          System.out.println(count++);
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    String message = new String(delivery.getBody());
    System.out.println(" [x] Received '" + message + "'");
  }

【问题讨论】：

您使用自动确认，它会在消费者请求消息后立即从队列中删除消息，无论它是否被正确处理，是否已传递给客户端。尝试使用手动消息确认。

标签： queue cloud rabbitmq bigdata

【解决方案1】：

因此，您的方案面临的挑战是您如何处理确认。

channel.basicConsume(QUEUE_NAME, true, consumer);

是问题所在。 true 的第二个参数是自动确认字段。

要解决这个问题，请使用：

channel.basicConsume(QUEUE_NAME, false, consumer);
while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    //...
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}

看起来您正在使用 RabbitMQ 的教程，并且您的代码 sn-p 来自 part one。如果您查看part two，他们会开始谈论确认和设置服务质量以提供循环调度。

值得指出的是basicConsume() 和nextDelivery() 的组合依赖于消费者内部的隐藏队列。因此，当您调用basicConsume() 时，几条消息被拉到客户端到本地存储。

这种方法的好处是它避免了因调用每条消息而产生的额外网络开销。问题是它可以在您的本地消费者中放置比您希望的更多的消息，如果消费者在处理本地隐藏队列中的所有消息之前退出，您可能会丢失消息。

如果您真的希望您的消费者一次只处理一条消息，以免丢失任何内容，您可能希望查看basicGet() 方法而不是basicConsume()。

【讨论】：