RabbitMQ 入门指南

发表于 2024/11/29

作者 Jasonbourne 38 分钟阅读

RabbitMQ是目前非常热门的一款消息中间件，其凭借高可靠、易扩展、高可用及丰富的功能特性受到众多开发者的青睐，在互联网、金融传统行业中都在大量的使用。

消息中间件

什么是消息中间件

消息中间件，也称消息队列，是一种在分布式系统中用于消息通信、传输和处理的软件，它提供了异步通信的机制，解耦了系统中的不同服务或组件，使它们能够以松耦合的方式进行交互。

它一般有两种传递模式：点对点模式和发布/订阅模式。点对点模式基于队列，消息生产者将消息发送至队列，消费者从队列中接受消息，队列的存在实现了消息的异步传输。发布/订阅模式基于主题，主题可以认为是消息传递的中介，消息发布者将消息发布到某个主题，消息订阅者从主题中订阅消息。发布/订阅模式在消息的一对多广播时采用。

消息中间件的作用

解耦：上游服务只需要将事件发布到消息中间件，降低了对下游服务的依赖。通过在系统中增加中间层（消息中间件），实现了业务间的解耦。
削峰填谷：消息中间件可以对瞬发的流量洪峰进行缓冲，让后端服务从容的应对流量压力。
异步通信：通过将事件发布到消息中间件实现异步处理，进而提升服务的吞吐量。

开源消息中间件对比

名称	吞吐量	集群支持	多客户端支持	扩展性	特点	典型应用场景
ActiveMQ	低（几千到万级消息/秒）	支持	支持大部分主流语言	差	无	传统应用系统、企业内部集成
RabbitMQ	中等（万级消息/秒）	支持	支持大部分主流语言	一般	支持AMQP,多租户，界面友好	短信、推送、异步任务处理
RocketMQ	高（十万级消息/秒）	支持	java语言	强，支持水平扩展	分布式事务	金融、电商、分布式事务、数据传输
Kafka	极高（百万级消息/秒）	支持	支持大部分主流语言	强，支持水平扩展	回溯消费,主题分区多副本,数据持久化保存	大数据处理、日志收集、实时分析

RabbitMQ 简介

RabbitMQ 是基于 Erlang 语言实现 AMQP(高级消息队列协议) 的消息中间件，它最初起源于金融系统，用于分布式系统中存储转发消息。它的主要特点如下：

可靠性：通过持久化、发布确认、传输确认等机制保证消息可靠性。
灵活的路由：通过不同模式的交换器，可以实现多种消息路由方式。
扩展性：可以通过部署多个 RabbitMQ 节点组建集群，并可以动态扩展集群节点。
高可用性：队列可以在集群的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。
多种协议：RabbitMQ 除了原生支持 AMQP 协议，还支持 STOMP、MQTT 等多种消息中间件协议。
多语言客户端：RabbitMQ 几乎支持所有常用语言，比如 Java、C#、Go、Python等。
管理界面：RabbitMQ 提供了一个易用的用户界面，方便用户对消息进行管理和监控。
插件机制：RabbitMQ 提供了许多插件，以实现多方面扩展。

使用 Docker 部署 RabbitMQ

Rabbit镜像仓库：https://hub.docker.com/_/rabbitmq

在装有 Docker 的环境下，执行以下命令安装 RabbitMQ：

  
docker run -d --name rabbit1 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 5672:5672 -p 15672:15672  rabbitmq:3-management

name：指定容器名称
RABBITMQ_DEFAULT_USER：默认用户
RABBITMQ_DEFAULT_PASS：默认密码
-p 5672:5672：服务默认侦听端口 5672
-p 15672:15672：web 管理页面端口 15672

执行成功后，访问 http://localhost:15672，即可进入 RabbitMQ 管理界面。

RabbitMQ 基础概念

生产者与消费者

生产者(Producer)

生产者创建消息，然后发布到 RabbitMQ 中。消息一般可以包含 2 个部分:消息体和标签。消息体也可以称之为 payload，在实际应用中，消息体一般是一个带有业务逻辑结构的数据，比如一个 JSON 字符串。当然可以进一步对这个消息体进行序列化操作。消息的标签用来表述这条消息，比如一个交换器的名称和一个路由键。生产者把消息交由 RabbitMQ， RabbitMQ 之后会根据标签把消息发送给相关的消费者。

消费者(Consumer)

消费者连接到 RabbitMQ 服务器，并订阅到队列上。当消费者消费一条消息时，只是消费消息的消息体 Cpayload)。在消息路由的过程中，消息的标签会丢弃，存入到队列中的消息只有消息体，消费者也只会消费到消息体，也就不知道消息的生产者是谁，当然消费者也不需要知道。

Broker

对于 RabbitMQ 来说，一个 RabbitMQ Broker 可以简单地看作一个 RabbitMQ 服务节点，或者 RabbitMQ 服务实例。大多数情况下也可以将一个 RabbitMQ Broker 看作一台 RabbitMQ 服务器。

队列与交换器

队列(queue)

队列是 RabbitMQ 的内部对象，用于存储消息。RabbitMQ 的消息只能存储在队列中。消费者可以从队列中获取消息并消费。如多个消费者订阅同一个队列，这时队列内的消息会被平均分摊（轮询）给多个消费者处理。

交换器(Exchange)

RabbitMQ 中，生产者会将消息先发送到交换器，然后由交换器根据路由规则将消息转发到队列中，如果路由不到，或许会返回给生产者，或许直接丢弃。

交换器有四种类型：

fanout：会将所有发送到该交换器的消息路由到与该交换器绑定的队列中，不判断 RoutingKey。
direct：会将消息路由到那些 BindingKey 和 RoutingKey 完全匹配的队列中。
topic：与 direct 类型的交换器相似，也是将消息路由到 BindingKey 和 RoutingKey 相匹配的队列中，但这里的匹配规则有些不同，它约定:
- RoutingKey 为一个点号”. “分隔的字符串(被点号” “分隔开的每一段独立的字符串称为一个单词 )，如“com.rabbitmq.client”;　　
- BindingKey 和 RoutingKey 一样也是点号”. “分隔的字符串;
- BindingKey 中可以存在两种特殊字符串”“和”#”，用于做模糊匹配，其中”“用于匹配一个单词，”#”用于匹配多规格单词(可以是零个)。
headers：header 类型的交换器不依赖于路由键的匹配规则路由消息，而是根据消息内容中的 header 属性进行匹配。该类型的交换器性能很差，而且也不实用，基本上不会看到它的存在。

绑定(Binding)

RabbitMQ 中通过绑定将交换器与队列关联起来，在绑定的时候一般会指定一个绑定键 (BindingKey)，这样 RabbitMQ 就知道如何正确地将消息路由到队列了。

路由键(routingkey)

生产者将消息发送给交换器时，一般会指定一个routingkey，用来指定消息的路由规则， routingkey需要与交换器类型和绑定键 (BindingKey) 联合使用才能最终生效。

Connection 与 Channel

无论是生产者还是消费者，都需要与 RabbitMQ Broker 建立连接，这个连接是一条 TCP 连接，也就是 Connection。很多时候应用程序需要多线程从 RabbitMQ 消费消息或生产消息，那么就需要创建多个 Connection，也就是多个 TCP 连接，但是保留许多的 TCP 连接是不可取的，因为这样做会大量消耗系统资源。因此 AMQP 0-9-1 连接采用了多路复用模式，可以认为是“轻量级的共享单个 TCP 连接的连接”，在单个连接上创建多条 channel。

客户端执行的每个协议操作都发生在一个 Channel 上。特定 Channel 上的通信是完全独立的来自另一个通道上的通信，因此每个协议 method 还带有一个 channelID，一个整数代理和客户端都使用它来确定该方法适用于哪个 Channel。Channel 仅存在于连接的上下文中，从不单独存在。当连接关闭时，该连接上的所有 Channel 也会关闭。对于使用多个线程/进程的应用程序处理时，为每个线程/进程打开一个新通道是很常见的并且不在它们之间共享 Channel。

Go开发示例

生产端示例

  
package main

import (
        "context"
        "log"
        "os"
        "strings"
        "time"

        amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
        if err != nil {
                log.Panicf("%s: %s", msg, err)
        }
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()

        ch, err := conn.Channel()
        failOnError(err, "Failed to open a channel")
        defer ch.Close()

        err = ch.ExchangeDeclare(
                "logs",   // name
                "fanout", // type
                true,     // durable
                false,    // auto-deleted
                false,    // internal
                false,    // no-wait
                nil,      // arguments
        )
        failOnError(err, "Failed to declare an exchange")

        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        body := bodyFrom(os.Args)
        err = ch.PublishWithContext(ctx,
                "logs", // exchange
                "",     // routing key
                false,  // mandatory
                false,  // immediate
                amqp.Publishing{
                        ContentType: "text/plain",
                        Body:        []byte(body),
                })
        failOnError(err, "Failed to publish a message")

        log.Printf(" [x] Sent %s", body)
}

func bodyFrom(args []string) string {
        var s string
        if (len(args) < 2) || os.Args[1] == "" {
                s = "hello"
        } else {
                s = strings.Join(args[1:], " ")
        }
        return s
}

消费端示例

  
package main

import (
        "log"

        amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
        if err != nil {
                log.Panicf("%s: %s", msg, err)
        }
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()

        ch, err := conn.Channel()
        failOnError(err, "Failed to open a channel")
        defer ch.Close()

        err = ch.ExchangeDeclare(
                "logs",   // name
                "fanout", // type
                true,     // durable
                false,    // auto-deleted
                false,    // internal
                false,    // no-wait
                nil,      // arguments
        )
        failOnError(err, "Failed to declare an exchange")

        q, err := ch.QueueDeclare(
                "",    // name
                false, // durable
                false, // delete when unused
                true,  // exclusive
                false, // no-wait
                nil,   // arguments
        )
        failOnError(err, "Failed to declare a queue")

        err = ch.QueueBind(
                q.Name, // queue name
                "",     // routing key
                "logs", // exchange
                false,
                nil,
        )
        failOnError(err, "Failed to bind a queue")

        msgs, err := ch.Consume(
                q.Name, // queue
                "",     // consumer
                true,   // auto-ack
                false,  // exclusive
                false,  // no-local
                false,  // no-wait
                nil,    // args
        )
        failOnError(err, "Failed to register a consumer")

        var forever chan struct{}

        go func() {
                for d := range msgs {
                        log.Printf(" [x] %s", d.Body)
                }
        }()

        log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
        <-forever
}

可靠性与数据安全性

发生故障的原因

基于消息传递的系统通常是分布式的，因此网络连接问题和网络拥塞是最常见的故障类型。除了网络中断外，防火墙也可能会中断长时间空闲的连接，而网络故障的检测通常需要一定的时间。

除了网络问题，服务器和客户端应用程序也可能随时发生硬件故障，或因软件崩溃而停止工作。此外，即使客户端程序仍在运行，逻辑错误也可能导致通道或连接异常，迫使客户端重新建立连接或通道，以恢复正常运行。

网络连接故障与恢复

当客户端与 RabbitMQ 节点之间的网络连接发生故障时，客户端需要重新建立与 Broker 的连接，之前的连接将自动关闭，并需要重新打开。通常，连接失败时，客户端会抛出 Connection 异常（或类似的语言特性）来通知应用程序。大多数客户端库都提供自动连接恢复功能。但在某些情况下，这种自动恢复可能不适用。此时，开发人员可以通过定义自定义的连接故障事件处理程序来实现自己的恢复逻辑。

消息传输中的数据丢失与确认机制

当连接中断时，消息可能正处于以下阶段：

在客户端或服务器端进行编码/解码。
位于 TCP 缓冲区中。
在网络上传输。

在这种情况下，正在传输中的消息无法被成功投递，因此需要重新传输。为确保消息的可靠交付，RabbitMQ 提供了消息确认机制，用于通知服务器和客户端何时需要重新传输消息。确认类型：

消费者确认（Consumer Acknowledgment）：消费者向服务器确认消息已被接收和/或处理。
生产者确认（Publisher Acknowledgment）：服务器向生产者确认消息已成功传递到队列中。

尽管 TCP 能够确保数据包被传输到对端并在网络层面重传丢失的数据包，但它仅解决网络故障，而消息确认解决的是应用层的问题。消费者确认表示消息已被接收并转移了所有权，接收方需要对消息负责。因此，在应用程序完成所需的所有操作之前（如记录数据、转发或处理消息），不应发送确认。一旦消息被确认，RabbitMQ 会将该消息标记为已删除。

确认机制的可靠性

使用消息确认机制可以确保至少一次投递。如果不启用确认，消息在发布或消费的过程中可能会丢失，而此时只能提供最多一次投递的保证。

检测 TCP 连接中断

在某些网络故障中，丢失的数据包可能导致 TCP 连接中断，但操作系统可能需要较长时间（例如，在默认的 Linux 配置下，大约 11 分钟）才能检测到这种中断。为了更快地检测连接异常，AMQP 0-9-1 协议提供了心跳机制，通过发送定期的心跳包来及时检测连接中断或对端节点无响应。此外，心跳机制还能防止某些网络设备（如防火墙）因连接空闲而中断 TCP 连接。有关详细信息，请参阅 Heartbeats 指南。

RabbitMQ 端的数据安全

为了避免在 RabbitMQ 端（而非应用程序端）丢失消息，队列和消息必须能够应对节点重启、节点故障和硬件故障。

使用 RabbitMQ 支持的消息协议，应用程序可以控制队列和消息的持久性。因此，重要数据应使用持久队列（或下面介绍的复制队列），并确保消息由生产者发布为持久消息。

集群与队列内容复制

节点集群提供冗余能力，允许系统在单个节点发生故障时继续运行。在 RabbitMQ 集群中，所有定义（如交换器、绑定、用户等）都在整个集群中共享。 Quorum 队列、Stream 和 Super Stream（分区流）是支持数据复制的结构。 Leader 副本由一个节点托管，其余节点托管 Follower 副本。如果 Leader 副本故障，Follower 副本之一将被选为新的 Leader。队列状态的变更（如消息入队、投递、确认）通常由 Leader 副本处理，部分操作也可能在 Follower 上执行。无论 Leader 副本位于哪个节点，队列和流对外都保持可见且可访问。在 Leader 切换期间，Quorum 队列的消息传递会暂停，直到新的 Leader 选出，此过程对客户端透明。

独占队列与其连接的生命周期相关联，不能被复制，并且在节点重启后不会继续存在。

连接到故障节点的消费者需要重新连接。当新的 Leader 被选出后，消费者会自动恢复，无需手动执行重连或重新订阅操作。

生产者端的数据安全

当启用消息确认（Confirm）时，若生产者的通道或连接失败，生产者应重新发送尚未被 RabbitMQ 确认的消息。这可能导致消息重复投递，因为代理可能已发送确认但由于网络问题未到达生产者。因此，消费者应用程序应支持幂等性或实现去重机制。

确保消息已路由在某些场景下，生产者需要确保消息已成功路由到目标队列。

简单场景：生产者可以声明一个目标队列并直接将消息发布到该队列。
复杂场景：如果生产者需要知道消息是否至少路由到一个队列，可以使用 mandatory 标志。在这种情况下，若消息未路由到任何队列，RabbitMQ 会通过 basic.return 方法将回复代码和解释文本发送回生产者。
当发布消息到集群节点时，如果目标队列的副本之间出现流控制或网络故障，可能会影响消息的路由和存储。请参阅节点间心跳指南以获取更多信息。

消费者端的数据安全

在发生网络或节点故障时，消息可能会被重新投递，消费者必须准备好处理重复投递的消息。

建议消费者实现幂等性逻辑，或在应用程序中显式去重。当消息被重新排队后再次投递时，redelivered 标志会被设置。这表明消费者可能已经接收过该消息，但不能完全保证，因为消息可能因网络或消费者故障未被处理。如果未设置 redelivered 标志，则可以保证该消息未被处理过。

若去重成本较高，消费者可以仅对标记为 redelivered 的消息进行去重或幂等处理。

无法处理的消息

如果消费者无法处理某条消息，可以使用以下方法：

basic.reject：拒绝该消息并选择是否重新排队。
basic.nack：批量拒绝一组消息，同样可选择是否重新排队。
如果消息未被重新排队，RabbitMQ 可以根据配置将其发送到死信队列（DLQ）。

消费者取消通知

当消费者正在使用的队列被删除时，RabbitMQ 会发送取消通知。消费者应采取适当的恢复措施，例如：

从另一个队列中消费消息。
或在安全的情况下重新声明被删除的队列并恢复消费。

使用 Docker-Compose 部署 RabbitMQ 集群

首先创建一个自定义的 Docker 网络，以便三个节点可以相互通信。

docker network create rabbitmq-network

新建一个 docker-compose.yaml 文件

  
version: '3.8'
services:
  rabbitmq-node1:
    image: rabbitmq:3-management
    hostname: rabbitmq-node1
    container_name: rabbitmq-node1
    ports:
      - "15672:15672"  # RabbitMQ管理界面
      - "5672:5672"    # AMQP协议端口
    environment:
      RABBITMQ_ERLANG_COOKIE: "rabbitmq_cookie_secret" # 两个节点之间通信的凭据，必须一致
      RABBITMQ_NODENAME: "rabbit@rabbitmq-node1"
    networks:
      - rabbitmq-network

  rabbitmq-node2:
    image: rabbitmq:3-management
    hostname: rabbitmq-node2
    container_name: rabbitmq-node2
    ports:
      - "15673:15672"  # 第二个节点的管理界面
      - "5673:5672"    # 第二个节点的 AMQP 协议端口
    environment:
      RABBITMQ_ERLANG_COOKIE: "rabbitmq_cookie_secret"
      RABBITMQ_NODENAME: "rabbit@rabbitmq-node2"
    networks:
      - rabbitmq-network

  rabbitmq-node3:
    image: rabbitmq:3-management
    hostname: rabbitmq-node3
    container_name: rabbitmq-node3
    ports:
      - "15674:15672"  # 第二个节点的管理界面
      - "5674:5672"    # 第二个节点的 AMQP 协议端口
    environment:
      RABBITMQ_ERLANG_COOKIE: "rabbitmq_cookie_secret"
      RABBITMQ_NODENAME: "rabbit@rabbitmq-node3"
    networks:
      - rabbitmq-network

networks:
  rabbitmq-network:
    external: true

在 docker-compose.yml 文件所在目录中运行以下命令启动三节点集群：

docker-compose up -d

进入第一个节点的容器：

docker exec -it rabbitmq-node1 bash

使用 RabbitMQ CLI 将 rabbitmq-node2、rabbitmq-node3 加入集群：

rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@rabbitmq-node2
rabbitmqctl join_cluster rabbit@rabbitmq-node3
rabbitmqctl start_app

验证集群状态

rabbitmqctl cluster_status

传统队列、镜像队列、Quorum队列对比

RabbitMQ 集群中，传统队列、镜像队列和 quorum 队列有不同的实现和特性，它们的主要区别体现在高可用性、容错性和数据一致性等方面。下面详细解释它们的区别：

传统队列（Classic Queue）

传统队列是 RabbitMQ 最初的队列类型，它有以下特点：

单点存储：消息只存储在单一的队列节点上。即使队列属于集群中的某个节点，消息也仅存在于该节点。
无容错机制：如果队列所在的节点出现故障，队列中的消息会丢失，除非启用了消息持久化。此时消费者可能会丢失未被消费的消息。
消息持久化：通过设置队列和消息为持久化（durable 和 persistent），可以确保即使 RabbitMQ 节点重启，消息也不会丢失。然而，持久化消息仍然只存在于单一节点，可能会导致单点故障。

适用场景：

对消息丢失有一定容忍度，或可以容忍短期停机的情况。
适用于对高可用性和容错性要求较低的场景。

镜像队列（Mirrored Queue）

镜像队列是在 RabbitMQ 中用于提高队列高可用性的队列类型，其特点如下：

队列副本：镜像队列的消息会同步到集群中其他节点上，消息会有一个或多个副本（mirrored queues）。通常，队列会有一个主节点（master）和多个镜像节点（mirror）。消息不仅存储在主节点上，还会实时复制到其他节点。
高可用性：如果主节点发生故障，镜像队列会自动选择一个镜像节点作为新的主节点，从而避免消息丢失，保证消息的可用性。
镜像队列的管理：镜像队列需要配置在特定的队列上。例如，使用 x-ha-policy 来定义镜像策略。
1 rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'
这样会使得所有队列都有镜像副本。
性能开销：镜像队列的副本机制会增加存储和网络带宽开销，尤其是在有大量队列或高吞吐量的场景下，性能可能会受到影响。

适用场景：

需要高可用性和容错性，能够承受节点故障的场景。
比较简单、快速地实现队列高可用性的场景。

Quorum 队列

Quorum 队列是 RabbitMQ 在 3.8 版本后引入的一种新型队列类型，旨在提供更强的容错性和一致性。它基于 Raft 协议来实现队列和消息的高可用性。其特点如下：

分布式一致性：Quorum 队列通过 Raft 协议保证了分布式系统中的一致性。Raft 协议保证了在多个节点之间进行消息复制时的数据一致性。
集群容错性：Quorum 队列没有主从之分，而是通过多个副本进行消息复制。在集群中的多个节点之间存储副本，当某个节点失效时，队列可以继续从其他副本节点提供服务。队列可以在多个节点上分布，确保即使发生节点故障，也能保持数据的一致性和高可用性。
自动恢复：当节点或队列崩溃时，Quorum 队列会自动恢复，选举新的队列副本，并继续提供服务。这个机制基于 Raft 协议自动处理节点之间的领导者选举。
消息确认：Quorum 队列在消息确认时具有更高的一致性，通常会等待所有副本的确认才算成功。这样可以避免丢失消息，但会带来一定的性能开销。

适用场景：

需要强一致性和高容错性，尤其是涉及关键任务或不能丢失消息的应用。
适用于对消息丢失非常敏感的系统，比如金融系统、银行系统、订单管理等。
在高可用性要求高并且对性能要求不如镜像队列那么宽松的场景下使用。

三者的对比

特性	传统队列	镜像队列	Quorum 队列
数据存储方式	存储在单一节点	存储在多个节点的副本中	存储在多个节点中，基于 Raft 协议实现一致性
高可用性	单节点故障时可能丢失消息	自动容错，主节点故障时选举副本	高可用，基于 Raft 保证一致性和容错性
性能	最高，但不具备容错性	性能会受到副本同步的影响	性能相对较低，但保证高一致性和容错性
容错性	低	高，主节点失败时选举新主节点	极高，支持多个副本和一致性保证
配置复杂性	简单	中等，需要配置镜像策略	较复杂，基于 Raft 协议，适合高可靠性要求
适用场景	非关键性、容错要求不高的场景	高可用性要求但不需要强一致性的场景	高可用性、强一致性、高容错性要求的关键业务场景

总结：

传统队列：适合高吞吐量、低容错要求的应用，但对高可用性和一致性要求较高的场景不适合。
镜像队列：适合需要较高可用性、但对一致性要求不高的场景，能够在节点故障时保证消息不丢失。
Quorum 队列：适合需要强一致性和高容错性的场景，尤其是在金融、银行等关键任务系统中。它通过 Raft 协议提供了更高的数据一致性和故障恢复能力，适用于高可靠性和高一致性要求的应用。

消息队列

RabbitMQ

本文由作者按照 CC BY 4.0 进行授权

消息中间件

什么是消息中间件

消息中间件的作用

开源消息中间件对比

RabbitMQ 简介

使用 Docker 部署 RabbitMQ

RabbitMQ 基础概念

生产者与消费者

生产者(Producer)

消费者(Consumer)

Broker

队列与交换器

队列(queue)

交换器(Exchange)

绑定(Binding)

路由键(routingkey)

Connection 与 Channel

Go开发示例

生产端示例

消费端示例

可靠性与数据安全性

发生故障的原因

网络连接故障与恢复

消息传输中的数据丢失与确认机制

确认机制的可靠性

检测 TCP 连接中断

RabbitMQ 端的数据安全

集群与队列内容复制

生产者端的数据安全

消费者端的数据安全

无法处理的消息

消费者取消通知

使用 Docker-Compose 部署 RabbitMQ 集群

传统队列、镜像队列、Quorum队列 对比

传统队列（Classic Queue）

镜像队列（Mirrored Queue）

Quorum 队列

三者的对比

总结：

热门标签

传统队列、镜像队列、Quorum队列对比