Close RabbitMQ channels and connections properly

Question

I am currently running a basic RabbitMQ topic publishing that publishes every 3 seconds.

My class looks like this:

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();
    }
    //...
}

I expect that every time when the method publish runs, it will publish and then the channel and connection both close, preventing a new channel and connection from stuck in the memory every 3 seconds.

However, when I look at my RabbitMQ admin interface (at the overview page), the Global Counts section shows that both the total number of connections and channels keep on increasing.

Eventually, my application crashes due to Socket limit and Memory limit being reached.

So it seems like close() doesn't delete the channel and connection used and still keep them in memory that eventually cause all the memory being consumed. What is proper method to use on the channel and connection to make sure that they don't do this?

Solution

However, when I look at my RabbitMQ admin interface (at the overview page), the Global Counts section shows that both the total number of connections and channels keep on increasing.

I think you have some problem with your connections.

Check your TCP connections in your OS, for example:

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

check also your connections using the command line tool:

2. rabbitmqctl list_connections if you have lot of connections in 1 and 2 means that you are not closing the connections/channels in the right way.

If you need to handle lot of connection you could increase the file descriptors configuration:

for example https://www.rabbitmq.com/install-debian.html

With systemd (Recent Linux Distributions)

On distributions that use systemd, the OS limits are controlled via a configuration file at /etc/systemd/system/rabbitmq-server.service.d/limits.conf, for example:

[Service] LimitNOFILE=300000

EDIT

The code you posted is fine, close() method is the right way to close the connection.

You should investigate in your real code if you execute the code.

Check also inside the log if there is:

=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 is the close connection

Comments

[Arnaud]

Couldn't you re-use the same connection and channel instead of creating them each time ?

[OrangeDog]

You're supposed to keep a single channel open and use it for multiple messages.

[pike]

@Berger Hi, I originally was doing this using AMQP, which creates the connection and channel when the application first starts. However, my new requirement is that I have to be able to change the host, username, and password on the fly without having to stop the application.

[pike]

@OrangeDog Because of this limitation, I am explicitly creating the connection, so that at the time of creation I can set the localhost, username, and password for the connection without having to stop the application.

[OrangeDog]

You should just override the relevant factory and/or config rather than completely reimplementing spring-amqp.

[pike]

@OrangeDog This could be something that I have missed. Do you know where I can find examples on how to override the relevant factory? Thanks.

[OrangeDog]

@pike go and read the documentation, find the relevant factory and work it out. If you then have a specific question, ask it.

[ka3ak]

@pike Just want to mention that connection.close() closes "all its channels" too according to the javadoc.

[Greg]

Old post but you could take a look at 'hoplin.io' lib on how to use the client using 'try-with-resouces' pattern

[pike]

Thanks for the reply. Based on the checks through command line, I do have a lot of connections, which means I am not closing my connections properly. But I thought by using the method close(), I should expect the connection to close, no? I have also tried using the method abort(), but no luck either. Do you happen to know what code I should use to close the dangling connections? Thanks.

[Gajukorse]

Even I am facing same issue . Connection is not getting closed at all