为什么我的RabbitMQ频道会继续关闭？

Question

我正在调试一些使用Apache POI从Microsoft Office文档中提取数据的Java代码.有时,当内存不足时,它会遇到大文档和POI崩溃.此时,它会尝试将错误发布到RabbitMQ,以便其他组件可以知道此步骤失败并采取适当的操作.但是,当它尝试发布到队列时,它会得到一个com.rabbitmq.client.AlreadyClosedException (clean connection shutdown; reason: Attempt to use closed channel).

这是错误处理程序代码:

try {
    //Extraction and indexing code
}
catch(Throwable t) {
    // Something went wrong! We'll publish the error and then move on with
    // our lives
    System.out.println("Error received when indexing message: ");
    t.printStackTrace();
    System.out.println();
    String error = PrintExc.format(t);
    message.put("error", error);

    if(mime == null) {
        mime = "application/vnd.unknown";
    }

    message.put("mime", mime);
    publish("IndexFailure", "", MessageProperties.PERSISTENT_BASIC, message);
}

为了完整性,这是发布方法:

private void publish(String exch, String route, 
    AMQP.BasicProperties props, Map<String, Object> message) throws Exception{
    chan.basicPublish(exch, route, props, 
        JSONValue.toJSONString(message).getBytes());  
}

我在try块中找不到任何似乎关闭RabbitMQ通道的代码.是否存在可以隐式关闭频道的情况？

编辑:我应该注意,basicPublish发布内部的调用抛出了AlreadyClosedException .

Answer 1

AMQP通道在通道错误时关闭.两个可能导致通道错误的常见问题:

• 尝试将消息发布到不存在的交换
• 尝试发布具有立即标志集的消息,该消息没有具有活动使用者集的队列

我将研究在您尝试使用addShutdownListener()发布消息的通道上设置ShutdownListener 以捕获关闭事件并查看导致它的原因.

Answer 2

在我的情况下另一个原因是我错误地确认了两次消息.这会在第二次确认后导致日志中出现RabbitMQ错误.

=ERROR REPORT==== 11-Dec-2012::09:48:29 ===
connection <0.6792.0>, channel 1 - error:
{amqp_error,precondition_failed,"unknown delivery tag 1",'basic.ack'}

删除重复的确认后,错误消失,通道不再关闭,AlreadyClosedException也消失了.

Answer 3

显然，AMQP 连接和/或通道突然关闭的原因有很多。就我而言，队列中有太多未确认的消息，因为消费者没有指定 prefetch_count ，因此连接每约 1 分钟就会终止一次。通过将消费者的预取计数设置为非零值来限制未确认消息的数量解决了该问题。

channel.basicQos(100); 

Answer 4

我想为要搜索该主题的其他用户添加此信息

接收通道关闭异常的另一个可能原因是发布者和使用者使用不同的队列声明/设置访问通道/队列时

发行人

channel.queueDeclare("task_queue", durable, false, false, null);

工人

channel.queueDeclare("task_queue", false, false, false, null);

从RabbitMQ网站

RabbitMQ doesn't allow you to redefine an existing queue with different parameters and will return an error to any program that tries to do that