RabbitMQ与通道Java线程安全

Question

我有一个从Rabbit接收消息的应用程序。当收到一条消息时，它会对它进行处理，然后在完成时执行ACK。应用程序可以在一个固定的线程池中同时处理2个项目，有2个线程。Rabbit的QOS预取设置为2，因为我不想在一个时间框架内给应用提供超过它所能处理的内容。

现在，我的消费者的handleDelivery执行以下操作：

Task run = new Task(JSON.parse(message));    
service.execute(new TestWrapperThread(getChannel(),run,envelope.getDeliveryTag()));

此时，您已经发现TestWrapperThread将channel.basicack（deliveryTag,false）；调用作为最后一个操作。

根据我对文档的理解，这是不正确的，而且可能有害，因为通道不是线程安全的，而且这种行为可能会把事情搞砸。那我该怎么做呢？我的意思是，我有一些想法，但他们会让一切变得更复杂，我想弄清楚它是否真的有必要。

提前致谢

Answer 1

我想您使用channel只是为了您的消费者，而不是其他操作，如发布等。

在您的情况下,唯一的潜在问题是：

channel.basicAck(deliveryTag, false);

因为您跨两个线程调用此操作，顺便说一句，如果您看到java代码，则此操作是安全的：

public void basicAck(long deliveryTag, boolean multiple)
   throws IOException
{
   transmit(new Basic.Ack(deliveryTag, multiple));
}

public void transmit(Method m) throws IOException {
   synchronized (_channelMutex) {
       transmit(new AMQCommand(m));
   }
}

_channelmutex是受保护的最终对象_channelmutex=new Object（）；

类创建的。请参阅amqchannel.java的github代码

编辑

希望有帮助。

编辑2我还加上尼古拉斯的评论：

请注意，从多个线程消费（basicConsumer）和添加是java客户机已经使用的常见rabbitmq模式。