RabbitMQ 中的事务消费答案

【问题标题】：Transactional consumption in RabbitMQRabbitMQ 中的事务消费
【发布时间】：2017-08-01 12:53:35
【问题描述】：

我还是 RabbitMQ 的新手。为了这些目的，我需要一个 MOM 系统：

在执行我的逻辑之前，会使用已发布的消息成功。
代理不必从排队，直到我的逻辑成功执行。

为了这些目标，我在第一次尝试时编写了以下代码：

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    boolean autoAck = false;
    channel.basicConsume(QUEUE_NAME, autoAck, "ProcessorOneTag",
            new DefaultConsumer(channel)
            {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)throws IOException
                {
                    try
                    {
                        channel.txSelect();
                        String routingKey = envelope.getRoutingKey();
                        String contentType = properties.getContentType();
                        long deliveryTag = envelope.getDeliveryTag();
                        System.out.println("Recieve Message is :" + new String(body));
                        int reslt = //execute my logic   
                        if(result == 0)
                            channel.txCommit();
                        else
                            channel.txRollback();
                    }
                    catch(Throwable t)
                    {
                        t.printStackTrace();
                    }
                }
            });

通过这种方法，我达到了第二个目的，换句话说，代理不会删除我的消息，但是有一次队列中的所有消息都被消耗并且所有消息都回滚了，代理不会将消息发送到又是我的消费者。

在第二次尝试时，我写了以下代码：

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    boolean autoAck = false;
    channel.basicConsume(QUEUE_NAME, autoAck, "ProcessorOneTag",
            new DefaultConsumer(channel)
            {
                @Override
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body)throws IOException
                {
                    try
                    {
                        String routingKey = envelope.getRoutingKey();
                        String contentType = properties.getContentType();
                        long deliveryTag = envelope.getDeliveryTag();
                        System.out.println("Recieve Message is :" + new String(body));
                        int reslt = //execute my logic   
                        if(result == 0)
                            channel.basicAck(deliveryTag, false);
                        else
                            channel.basicNack(deliveryTag,false,true);
                    }
                    catch(Throwable t)
                    {
                        t.printStackTrace();
                    }
                }
            });

通过这个解决方案，我实现了两个目标，但我不知道我的代码是否正确？该方法是否会在高 TPS 的生产环境中引起问题？不知道basicNack方法的requeue标志是重还是轻？

【问题讨论】：

我使用basicAck和basicNack已经很久了，即使TPS很高也没有问题。

标签： java rabbitmq jms mom

【解决方案1】：

几个月前我也有同样的要求。我经历了许多解决方案，这对我有用。

我将传递标签存储在内存中的某处，如果我的逻辑运行良好，我会手动确认消息或拒绝该消息。为此，我使用了以下方法。

if (success)
connectionModel.BasicAck(Convert.ToUInt64(uTag), false);
else
connectionModel.BasicReject(Convert.ToUInt64(uTag), true);

以上代码在我的生产中运行良好，速度为 5000msg/sec。好吧，basicNack 方法给我带来了问题。

【讨论】：