正确关闭RabbitMQ通道和连接

Question

我目前正在运行一个基本的RabbitMQ主题发布程序，每3秒钟发布一次。

我的类看起来像这样：

import com.rabbitmq.client.*;

import java.io.IOException;

public class EmitLogTopic {

    private static final String EXCHANGE_NAME = "topic_logs";

    @Scheduled(fixedRate = 3000)
    public void publish(String[] argv)
                  throws Exception {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String routingKey = getRouting(argv);
        String message = getMessage(argv);

        channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
        System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");

        channel.close();
        connection.close();
    }
    //...
}

我希望每次运行“publish”方法时，它都会发布消息，然后关闭通道和连接，以防止新的连接和通道每个3秒钟被卡在内存中。

但是，当我查看RabbitMQ管理页面（概述页面），全局计数部分显示连接和通道的总数不断增加。

最终，我的应用程序因达到Socket限制和内存限制而崩溃。

因此，似乎close()不能删除使用的通道和连接，而是将它们保留在内存中，最终导致所有内存被消耗。有什么适当的方法可以确保它们不会这样做吗？

- pike

---

你应该只覆盖相关的工厂和/或配置，而不是完全重新实现 spring-amqp。 - OrangeDog

@OrangeDog 这可能是我错过的东西。你知道在哪里可以找到有关如何覆盖相关工厂的示例吗？谢谢。 - pike

@pike 去阅读文档，找到相关的工厂并解决问题。如果你有具体的问题，请提出来。 - OrangeDog

@pike 只是想提醒一下，connection.close() 根据 javadoc 的说明也会关闭 "所有的通道"。 - ka3ak

旧帖子，但你可以看一下 'hoplin.io' 库，了解如何使用“try-with-resources”模式来使用客户端。 - Greg

Answer 1

然而，当我查看我的RabbitMQ管理界面（概述页面）时，全局计数部分显示总连接数和通道数都在不断增加。

我认为您的连接存在问题。

检查您的操作系统中的TCP连接，例如：

1. netstat -anp | grep :5672 | grep ESTABLISHED | wc -l

还可以使用命令行工具检查您的连接：

2. rabbitmqctl list_connections 如果在1和2中有很多连接，则意味着您没有以正确的方式关闭连接/通道。

如果您需要处理大量连接，则可以增加文件描述符配置：

例如https://www.rabbitmq.com/install-debian.html 使用systemd（最新的Linux发行版）

在使用systemd的发行版上，操作系统限制是通过位于/etc/systemd/system/rabbitmq-server.service.d/limits.conf的配置文件进行控制的，例如：

[Service] LimitNOFILE=300000 编辑 您发布的代码没问题，close()方法是关闭连接的正确方式。

您应该在实际代码中进行调查是否执行了该代码。

还要检查日志中是否有：

=INFO REPORT==== 22-Aug-2017::09:23:28 ===
connection <0.383.0> ([::1]:60590 -> [::1]:5672): user 'guest' authenticated and granted access to vhost '/'

=INFO REPORT==== 22-Aug-2017::09:23:37 ===
closing AMQP connection <0.383.0> ([::1]:60590 -> [::1]:5672, vhost: '/', user: 'guest')

closing AMQP connection 是关闭连接的意思

- Gabriele Santomaggio

---

谢谢回复。通过命令行检查，我发现有很多连接，这意味着我没有正确关闭我的连接。但是我认为使用close()方法应该可以关闭连接，不是吗？我也尝试过使用abort()方法，但也没有成功。你知道我应该使用什么代码来关闭这些悬空的连接吗？谢谢。 - pike

我也遇到了同样的问题。连接根本没有关闭。 - Gajukorse