Channels and Connections in RabbitMQ – RabbitMQ中的通道和连接

最后修改: 2022年 10月 18日

1. Introduction


In this quick tutorial, we’ll show how to use RabbitMQ’s APIs related to two core concepts: Connections and Channels.


2. RabbitMQ Quick Recap


RabbitMQ is a popular implementation of the AMQP (Advanced Messaging Queue Protocol), widely used by companies of all sizes to handle their messaging needs.

RabbitMQ 是 AMQP(高级消息队列协议)的流行实现,被各种规模的公司广泛用于处理其消息传递需求。

From an application point of view, we’re usually concerned with AMQP’s main entities: Virtual Hosts, Exchanges, and Queues. As we’ve already covered those concepts in earlier articles, here, we’ll focus on the details of two less-discussed concepts: Connections and Channels.


3. Connections


The first step a client must take to interact with a RabbitMQ broker is to establish a Connection. AMPQ is an application-level protocol, so this connection happens on top of a transport-level one. This can be a regular TCP connection or an encrypted one using TLS. The main role of a Connection is to provide a secure conduit through which a client can interact with a broker.

客户端与 RabbitMQ 代理交互的第一步是建立一个连接。 AMPQ 是一个应用级协议,因此该连接发生在一个传输级协议之上。这可以是一个普通的 TCP 连接,也可以是一个使用 TLS 的加密连接。连接的主要作用是提供一个安全的管道,客户可以通过这个管道与代理进行交互。

This means that during connection establishment, a client must supply valid credentials to the server. A server may support different credential types, including regular username/password, SASL, X.509 password, or any supported mechanism.


Besides security, the connection establishment phase is also responsible for negotiating some aspects of the AMPQ protocol. At this point, if the client and/or server cannot agree on the protocol version or a tuning parameter value, the connection won’t be established, and the transport level connection will be closed.


3.1. Creating Connections in Java Applications


When using Java, the standard way to communicate with a RabbitMQ browser is to use the amqp-client Java library. We can add this library to our project using adding the corresponding Maven dependency:

使用 Java 时,与 RabbitMQ 浏览器通信的标准方式是使用 amqp-client Java 库。我们可以通过添加相应的 Maven 依赖关系将该库添加到我们的项目中。


The latest version of this artifact is available on Maven Central.

该工件的最新版本可在Maven Central上找到。

This library uses the Factory pattern to create new connections. First, we create a new ConnectionFactory instance and set all parameters needed to create connections. At a minimum, this requires informing the address of the RabbitMQ host:

这个库使用Factory模式来创建新的连接。首先,我们创建一个新的ConnectionFactory实例并设置创建连接所需的所有参数。至少,这需要告知 RabbitMQ 主机的地址。

ConnectionFactory factory = new ConnectionFactory();

Once we’re done setting those parameters, we use the newConnection() factory method to create a new Connection instance:


Connection conn = factory.newConnection();

4. Channels


Simply put, an AMQP channel is a mechanism that allows multiplexing multiple logic flows on top of a single connection. This allows better resource usage both on the client and server-side since setting up a connection is a relatively expensive operation.


A client creates one or more channels so it can send commands to the broker. This includes commands related to sending and/or receiving messages.


Channels also provide some additional guarantees regarding the protocol logic:


  • Commands for a given channel are always executed in the same order they’re sent.
  • Given a scenario where a client opens multiple channels over a single connection, implementations can distribute the available bandwidth between them
  • Both parties can issue flow control commands, which inform the peer that it should stop sending messages.

A key aspect of a channel is that its lifecycle is bound to the connection used to create it. This means that if we close a connection, all associated channels will also be closed.


4.1. Creating Channels in Java Applications


Java applications using the amqp-client library create a new Channel from an existing Connection using the createChannel() method from the former:


channel = conn.createChannel();

Once we have a Channel, we can send commands to the server. For instance, to create a queue, we use the queueDeclare() method:


channel.queueDeclare("example.queue", true, false, true, null);

This code “declares” a queue, which is AMQP’s way of saying “create if not already existing”. The additional arguments after the queue name define its additional characteristics:

这段代码 “声明 “了一个队列,这是AMQP表示 “如果不存在就创建 “的方式。队列名称后面的额外参数定义了它的额外特性。

  • durable: this declaration is persistent, meaning it will survive a server’s restart
  • exclusive: this queue is restricted to the connection associated with the channel declaring it
  • autodelete: the server will delete the queue once no longer in use
  • args: optional map with arguments used to tune the queue behavior; for instance, we can use those arguments to define the TTL for messages and dead-letter behavior

Now, to post a message to this queue using the default exchange, we use the basicPublish() method:


channel.basicPublish("", queue, null, payload);

This code sends a message to the default exchange using the queue name as its routing key.


5. Channel Allocation Strategies


Let’s consider a scenario where we use messaging systems: CQRS (Command Query Responsibility Segregation) applications. In a nutshell, CQRS-based applications have two independent paths: commands and queries. Commands can change data but never return values. Queries, on the other hand, return values but never modify them.


Since the command path never returns any data, the service can execute them asynchronously. In a typical implementation, we have an HTTP POST endpoint that internally builds a message and sends it to a queue for later processing.

由于命令路径从不返回任何数据,服务可以异步地执行它们。在一个典型的实现中,我们有一个HTTP POST端点,它在内部建立了一个消息,并将其发送到一个队列中供以后处理。

Now, for a service that must handle dozens or even hundreds of concurrent requests, opening connections and channels every time is not a realistic option. Instead, a better approach is to use a channel pool.


Of course, this leads to the next problem: should we create a single connection and create channels from it or use multiple connections?


5.1. Single Connection/Multiple Channels


In this strategy, we’ll use a single connection and just create a channel pool with a capacity equal to the maximum number of concurrent connections the service can manage. For a traditional thread-per-request model, this should be set to the same size as the request handler thread pool.


The downside of this strategy is that, under heavier loads, the fact that we must send commands one at a time through the associated channel implies that we must use a synchronization mechanism. This, in turn, adds extra latency in the command path, which we want to minimize.


5.2. Connection-per-Thread Strategy


Another option is to go to the other extreme and use a Connection pool, so there’s never contention for a channel. For each Connection, we’ll create a single Channel that a handler thread will use to issue commands to the server.


However, the fact that we remove synchronization from the client side comes with a cost. The broker must allocate additional resources for each connection, such as socket descriptors and state information. Moreover, the server must split the available throughput between clients.


6. Benchmarking Strategies


To evaluate those candidate strategies, let’s run a simple benchmark for each one. The benchmark consists of running multiple workers in parallel that send one thousand messages of 4 Kbytes each. Upon construction, the worker receives a Connection from which it will create a Channel to send commands. It also receives the number of iterations, payload size, and a CountDownLatch used to inform the test runner that it has finished sending messages:


public class Worker implements Callable<Worker.WorkerResult> {
    // ... field and constructor omitted
    public WorkerResult call() throws Exception {

        try {
            long start = System.currentTimeMillis();
            for (int i = 0; i < iterations; i++) {
                channel.basicPublish("", queue, null, payload);

            long elapsed = System.currentTimeMillis() - start;
            return new WorkerResult(elapsed);
        } finally {
    public static class WorkerResult {
        public final long elapsed;

        WorkerResult(long elapsed) {
            this.elapsed = elapsed;

Besides indicating that it has finished its job by decrementing the latch, the worker also returns a WorkerResult instance with the elapsed time to send all messages. Although here we just have a long value, we can use extend it to return more details.


The controller creates the connection factory and workers according to the strategy being evaluated. For the single connection, it creates the Connection instance and passes it to every worker:


public Long call() {
    try {
        Connection connection = factory.newConnection();
        CountDownLatch counter = new CountDownLatch(workerCount);
        List<Worker> workers = new ArrayList<>();
        for( int i = 0 ; i < workerCount ; i++ ) {
            workers.add(new Worker("queue_" + i, connection, iterations, counter,payloadSize));

        ExecutorService executor = new ThreadPoolExecutor(workerCount, workerCount, 0,
          TimeUnit.SECONDS, new ArrayBlockingQueue<>(workerCount, true));
        long start = System.currentTimeMillis();
        if( counter.await(5, TimeUnit.MINUTES)) {
            long elapsed = System.currentTimeMillis() - start;
            return throughput(workerCount,iterations,elapsed);
        else {
            throw new RuntimeException("Timeout waiting workers to complete");
    catch(Exception ex) {
        throw new RuntimeException(ex);

For the multiple connections strategy, we create a new Connection for each worker:


for (int i = 0; i < workerCount; i++) {
    Connection conn = factory.newConnection();
    workers.add(new Worker("queue_" + i, conn, iterations, counter, payloadSize));

The throughput function calculates the benchmark measure will be the total time needed to complete all workers, divided by the number of workers:


private static long throughput(int workerCount, int iterations, long elapsed) {
    return (iterations * workerCount * 1000) / elapsed;

Notice that we need to multiply the numerator by 1000 so we get the throughput in messages by second.


7. Running the Benchmark


These are the results of our benchmark for both strategies. For each worker count, we’ve run the benchmark 10 times and used the average value as the throughput measure for tar particular worker/strategy. The environment itself is modest by today’s standards:


  • CPU: dual-core i7 dell notebook @ 3.0 GHz
  • Total RAM: 16 GB
  • RabbitMQ: 3.10.7 running on Docker (docker-machine with 4 GBytes RAM)


For this specific environment, we see a slight advantage for the single connection strategy. This advantage seems to increase for the 150 workers scenario.


8. Selecting a Strategy


Given the benchmark results, we cannot point to a clear winner. For worker counts between 5 and 100, the results are more or less the same. After that, the overhead associated with multiple connections seems to be higher than handling multiple channels on a single connection.


Also, we must consider that the test workers do only one thing: send fixed messages to a queue. Real-world applications, like the CQRS one we’ve mentioned, usually do some extra work before and/or after sending a message. So, to select the best strategy, the recommended way is to run your own benchmark using a configuration that is as close as possible to the production environment.


9. Conclusion


In this article, we’ve explored the concepts of Channels and Connections in RabbitMQ and how we can use them in different ways. As usual, the full code is available over on GitHub.

在这篇文章中,我们探讨了 RabbitMQ 中通道和连接的概念,以及我们如何以不同方式使用它们。像往常一样,完整的代码可在 GitHub 上获取