rabbitmq虚拟主机（rabbitmq主从搭建）

本文目录一览：

• 1、RabbitMQ架构
• 2、RabbitMQ集群
• 3、消息中间件——RabbitMQ（四）命令行与管控台的基本操作！

RabbitMQ架构

RabbitMQ是一个分布式系统

一、使用rabbitmq时的系统架构图

通过路由键将交换机和队列进行绑定，从而实现消息的发送和接收。

二、rabbitmq基本概念

rabbitmq是AMQP协议的一个开源实现，所以其内部实际上也是AMQP中的基本概念，如下图所示：

1、Message（消息）

消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（传输模式，指出该消息可能需要持久化存储）等。

2、Publisher

消息生产者，也是一个向交换器发布消息的客户端应用程序，就是投递消息的程序。

3、Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。消息交换机，它指定消息按什么规则，路由到哪个队列。

4、Routing Key

路由关键字，exchange根据这个关键字进行消息投递。

5、Binding（绑定）

用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

它的作用就是把exchange和queue按照路由规则绑定起来。

绑定其实就是关联了exchange和queue，或者这么说：queue对exchange的内容感兴趣，exchange要把它的Message deliver到queue。

6、Queue（消息队列）

消息的载体，每个消息都会被投到一个或多个队列，等待消费者连接到这个队列将其取走。它是消息的容器，也是消息的终点。

7、Connection

网络连接，例如一个TCP连接。

8、Channel（信道，通道）

消息通道，在客户端的每个连接里，可建立多个channel。

多路复用连接中的一条独立双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念以达到复用一条TCP连接的目的。

9、Consumer

消息消费者，表示一个从消息队列中取得消息的客户端应用程序，就是接受消息的程序。

10、Virtual Host

虚拟主机，表示一批交换器、消息队列和相关对象。一个broker里可以有多个vhost，用作不同用户的权限分离。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个vhost本质上就是一个mini版的rabbitmq服务器，拥有自己的队列、交换器、绑定和权限机制。

vhost是AMQP概念的基础，必须在连接时指定，rabbitmq默认的vhost是 / 。

11、Broker

表示消息队列服务器实体。它提供一种传输服务，它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输。

三、AMQP中的消息路由

生产者把消息发布到Exchange上，消息最终到达队列并被消费者接收，而Binding决定交换器的消息应该发送到那个队列。如下图所示：

四、Exchange类型

Exchange分发消息时根据类型的不同分发策略有区别，目前共有四种类型：direct、fanout、topic、headers。headers匹配AMQP消息的header而不是路由键，此外headers交换器和direct交换器完全一致，但性能差很多，目前几乎用不到。且看direct、fanout、topic这三种类型。

1、direct类型

消息中的路由键routing key如果和Binding中的binding key一致，交换器就将消息发到对应的队列中去。路由键与队列名完全匹配，如果一个队列绑定到交换器要求路由键为“dog”，则只转发routing key标记为“dog”的消息，不会转发“dog.puppy”等等。 它是完全匹配、单传播的模式。

Driect exchange的路由算法非常简单：通过bindingkey的完全匹配，可以用下图来说明：

Exchange和两个队列绑定在一起，Q1的bindingkey是orange，Q2的binding key是black和green。

当Producer publish key是orange时，exchange会把它放到Q1上，如果是black或green就会到Q2上，其余的Message被丢弃。

2、fanout类型

每个发到fanout类型交换器的消息都会分到所有绑定的队列上去。fanout交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。类似于子网广播，每台子网内的主机都获得了一份复制的消息。 fanout类型转发消息是最快的。  如下图所示：

3、topic类型

topic交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：#和*，#匹配0个或多个单词，*只能匹配一个单词。

对于Message的routing_key是有限制的，不能是任意的。格式是以点号“.”分割的字符表。比如：”stock.usd.nyse”，“nyse.vmw”， “quick.orange.rabbit”。你可以放任意的key在routing_key中，当然最长不能超过255 bytes。对于routing_key，有两个特殊字符#和*，#匹配0个或多个单词，*只能匹配一个单词。如下图所示：

Producer发送消息时需要设置routing_key，routing_key包含三个单词和两个点号，第一个key描述了celerity(灵巧)，第二个是color(色彩)，第三个是物种。

在这里我们创建了两个绑定： Q1 的binding key 是”.orange.“； Q2 是 “..rabbit” 和 “lazy.#”：Q1感兴趣所有orange颜色的动物；Q2感兴趣所有rabbits和所有的lazy的。

例如：rounting_key 为 “quick.orange.rabbit”将会发送到Q1和Q2中。rounting_key 为”lazy.orange.rabbit.hujj.ddd”会被投递到Q2中，#匹配0个或多个单词。

五、ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。

1、Connection

Connection是Rabbitmq的socket连接，它封装了socket协议相关部分逻辑。

2、ConnectionFactory

ConnectionFactory是connection的制造工厂。

3、Channel

Channel是我们与rabbitmq打交道的最重要的一个接口，大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

六、任务分发机制

1、Round-robin dispathching 循环分发

RabbbitMQ的分发机制非常适合扩展,而且它是专门为并发程序设计的,如果现在load加重,那么只需要创建更多的Consumer来进行任务处理。

2、Message acknowledgment 消息确认

为了保证数据不被丢失，RabbitMQ支持消息确认机制，为了保证数据能被正确处理而不仅仅是被Consumer收到，这就需要在处理完数据之后发送一个确认ack。

在处理完数据之后发送ack，就是告诉RabbitMQ数据已经被接收并且处理完成，RabbitMQ可以将消息从队列中移除了。如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer，这样就保证在Consumer异常退出情况下数据也不会丢失。

RabbitMQ没有用到超时机制，它仅仅通过Consumer的连接中断来确认该Message并没有被正确处理，一个消费者处理消息的时间再长也不会导致该消息被发送给其他消费者，即RabbitMQ给了Consumer足够长的时间来做数据处理。如果忘记ack，那么当Consumer退出时，Mesage会被重新分发，从而导致队列中的累积的消息越来越多，然后RabbitMQ会占用越来越多的内存。

3、Message durability 消息持久化

如果我们希望即使在rabbitmq服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置成可持久化的（durable），这样就可以保证绝大部分情况下我们的rabbitmq消息不会丢失。但依然解决不了小概率丢失事件的发生（例如rabbitmq服务器已经接收到了生产者的消息，但还没来得及持久化该消息时rabbitmq服务器就断电了）。如果也要将这种小概率事件管理起来就需要使用到事务了。要持久化队列需要在声明时指定durable=True；这里要注意,队列的名字一定要是Broker中不存在的，不然不能改变此队列的任何属性。队列和交换机有一个创建时候指定的标志durable，durable的唯一含义就是让具有这个标志的队列和交换机会在重启之后重新建立。

消息持久化包括3部分

（1）exchange持久化，在声明时指定durable = true

channel.ExchangeDeclare(ExchangeName,"direct", durable:true, autoDelete:false, arguments:null);//声明消息队列，且为可持久的

（2）queue持久化，在声明时指定durable = true

channel.QueueDeclare(QueueName, durable:true, exclusive:false, autoDelete:false, arguments:null);//声明消息队列，且为可持久的

（3）消息持久化，在投递时指定delivery_mode = 2(1是非持久化)。

channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());

如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的；如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定。

注意：一旦创建了队列和交换机，就不能修改其标志了。例如创建了一个non-durable的队列，然后想把它改变成durable的，唯一的办法就是删除这个队列然后重新创建。

4、Fair dispath 公平分发

你可能也注意到了，分发机制不是那么优雅，默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。n是取余后的，它不管Consumer是否还有unacked Message，只是按照这个默认的机制进行分发。那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却毫无休息的机会，那么Rabbit是如何处理这种问题呢?

通过basic.qos方法设置prefetch_count=1，如下设置

channel.basic_qos(prefetch_count=1)

这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message，换句话说，在接收到该Consumer的ack前，它不会将新的Message分发给它。但是这种方法可能会导致queue满。当然，这种情况下你可能需要添加更多的Consumer，或者创建更多的virtualHost来细化你的设计。

5、分发到多个Consumer

Direct Exchange：直接匹配，通过Exchange名称+RountingKey来发送与接收消息。

Fanout Exchange：广播订阅，向所有的消费者发布消息，但是只有消费者将队列绑定到该路由器才能收到消息，忽略Routing Key。

Topic Exchange：主题匹配订阅，这里的主题指的是RoutingKey，RoutingKey可以采用通配符，如：*或#，RoutingKey命名采用英文句点来分隔多个词，只有消息将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息。

Headers Exchange：消息头订阅，消息发布前为消息定义一个或多个键值对的消息头，然后消费者接收消息，同时需要定义类似的键值对请求头（如

x-mactch=all或者x_match=any），只有请求头与消息头匹配，才能接收消息,忽略RoutingKey。

默认的exchange：如果用空字符串去声明一个exchange，那么系统就会使用”amq.direct”这个exchange。我们创建一个queue时，默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去。如下：

channel.BasicPublish("","TaskQueue", properties, bytes);

因为在第一个参数选择了默认的exchange，而我们声明的队列叫TaskQueue，所以默认的，它要新建一个也叫TaskQueue的routingKey，并绑定在默认的exchange上，导致了我们可以在第二个参数routingKey中写TaskQueue，这样它就会找到定义的同名的queue并把消息放进去。

如果有两个接收程序都是用了同一个的queue和相同的routingKey去绑定direct exchange的话，分发的行为是负载均衡的，也就是说第一个是程序1收到，第二个是程序2收到，以此类推。

如果有两个接收程序用了各自的queue，但使用相同的routingKey去绑定direct exchange的话，分发的行为是复制的，即每个程序都会收到这个消息的副本。行为相当于fanout类型的exchange。

多个queue绑定同一个key也是可以的，对于下图的例子，Q1和Q2都绑定了black，对于routing key是black的Message，会被deliver到Q1和Q2，其余的Message都会被丢弃。

七、RPC远程过程调用

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。 但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

RabbitMQ中实现RPC的机制如下图所示：

客户端发送请求（消息）时，在消息的属性（MessageProperties ，在AMQP 协议中定义了14种properties ，这些属性会随着消息一起发送）中设置两个值replyTo （一个Queue 名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue 中）和correlationId （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行或执行失败）。

服务器端收到消息并处理，服务器端处理完消息后，将生成一条应答消息到replyTo 指定的Queue中 ，同时带上correlationId 属性，客户端之前已订阅replyTo 指定的Queue ，从中收到服务器的应答消息后，根据其中的correlationId 属性分析哪条请求被执行了，然后根据执行结果进行后续业务处理。

转发：

RabbitMQ集群

如果有一个消息生产者或者消息消费者通过amqp-client的客户端连接至节点1进行消息的发布或者订阅，那么此时的集群中的消息收发只与节点1相关。

如果消息生产者所连接的是节点2或者节点3，此时队列1的完整数据不在该两个节点上，那么在发送消息过程中这两个节点主要起了一个路由转发作用，根据这两个节点上的元数据转发至节点1上，最终发送的消息还是会存储至节点1的队列1上。

RabbitMQ 集群是一个或多个节点的逻辑分组，每个节点共享用户、虚拟主机、队列、交换器、绑定、运行时参数和其他分布式状态。

一些分布式系统 有leader和follower节点。 对于 RabbitMQ 来说， RabbitMQ集群中的所有节点都是平等的。

RabbitMQ 集群可以通过多种方式组成：

RabbitMQ 节点绑定到端口以接受客户端和 CLI 工具连接。其他进程和工具（例如 SELinux）可能会阻止 RabbitMQ 绑定到端口。发生这种情况时，节点将无法启动。

CLI工具、客户端库和 RabbitMQ 节点也会打开连接（客户端 TCP 套接字）。防火墙会阻止节点和 CLI 工具相互通信。确保可以访问以下端口：

RabbitMQ节点和 CLI 工具（例如 rabbitmqctl）使用 cookie 来确定它们是否被允许相互通信。为了让两个节点能够通信，它们必须具有相同的共享密钥，称为 Erlang cookie。cookie 只是一串最多 255 个字符的字母数字字符。 它通常存储在本地文件中。该文件必须只能由所有者访问（例如具有 600 或类似的 UNIX 权限）。每个集群节点必须具有相同的 cookie。

在 UNIX 系统上，cookie通常是位于/var/lib/rabbitmq/.erlang.cookie（由服务器使用）和$HOME/.erlang.cookie（由 CLI 工具使用）。

RabbitMQ 节点使用主机名相互寻址

!== 所有主机执行 ==

!== 所有主机执行 ==

!== 所有主机执行 ==

默认配置文件/usr/lib/rabbitmq/lib/rabbitmq_server-3.7.17/ebin/rabbit.app

!== node01主机执行 ==

!== node02主机执行 ==

!== node03主机执行 ==

!== 所有主机执行 ==

因RabbitMQ 集群元数据同步基于 cookie 共享方案实现

文件路径/var/lib/rabbitmq/.erlang.cookie

!== node02、node03主机执行 ==

!== 任意主机执行 ==

节点分为：磁盘节点及内存节点

RAM节点是一种特殊情况，可用于提高具有高队列、交换或绑定搅动的集群的性能。RAM节点不提供更高的消息速率。 官方建议在绝大多数情况下，仅使用磁盘节点。

如果一个集群中都是RAM节点，那么集群停止，将无法再次启动，并将丢失所有数据

官方提示：经典队列镜像将在未来版本中删除，考虑改用仲裁队列或非复制经典队列

每个镜像队列由一个领导副本和一个或多个镜像副本，leader被托管的节点成为leader节点。首先应用给定队列的所有操作 在队列的leader节点上，然后传播到镜像。

如果承载队列的leader节点出现故障，则只要已同步，最旧的镜像将升级为新的leader。根据队列镜像参数，也可以升级未同步的镜像。

队列通过策略启用了镜像，策略模式的说明如下：

每当队列的策略发生变化时，它都保持其现有的镜像尽可能适用新策略。

设置的镜像队列可以通过开启的网页的管理端Admin-Policies，也可以通过命令。

管理端界面：

命令行：

为避免集群中的某个节点托管大多数leader队列，因此导致高负载，leader队列应合理均匀的分布在集群节点上。控制leader队列分布节点策略有三种，可以在rabbitmq.conf文件中定义queue_master_locator参数设置

修改节点策略可能会导致现有的leader队列没有在新的策略中，为了防止消息丢失，RabbitMQ会保留现有的leader直到至少另一个镜像已同步，一旦发生同步，消费者将与leader断开连接，需要重新连接。

如果leader故障，其他镜像升级为leader过程如下：

如果消费者使用 自动确认模式 ，那么消息可能会丢失。这与非镜像队列没有什么不同：代理认为消息一旦以自动确认模式发送给消费者，就会被确认。

如果客户端突然断开连接，则可能永远不会收到消息。在镜像队列的情况下，如果leader死亡，那些正在以 自动确认模式 发送给消费者的消息可能永远不会被这些客户端接收，并且不会被新leader重新排队。由于消费客户端连接到存活节点的可能性，消费者取消通知有助于识别此类事件何时发生。 当然，在实践中，如果数据安全性不如吞吐量重要，那么自动确认模式是可行的方法。

节点可以随时加入集群。 根据队列的配置，当节点加入集群时，队列可能会在新节点上添加镜像。 此时，新镜像将是空的：它不会包含队列中任何现有的内容。 这样的镜像将接收发布到队列的新消息，因此随着时间的推移将准确地表示镜像队列的尾部。 随着消息从镜像队列中排出，新镜像丢失消息的队列头部的大小将缩小，直到最终镜像的内容与leader的内容完全匹配。 在这一点上，镜像可以被认为是完全同步的。

新添加的镜像不提供添加镜像之前存在的队列内容的额外形式的冗余或可用性，除非队列已明确同步。 由于在发生明确同步时队列变得无响应，因此最好允许正在从中排出消息的活动队列自然同步，并且仅明确同步非活动队列。

启用自动队列镜像时，请考虑所涉及队列的预期磁盘数据集 。 具有庞大数据集（例如，数十 GB 或更多）的队列必须将其复制到新添加的镜像中，这会给集群资源（例如网络带宽和磁盘 I/O）带来很大的负载。

要查看镜像状态（是否同步）：

手动同步队列：

取消正在进行的同步：

如果你停止一个包含镜像队列leader的 RabbitMQ 节点，其他节点上的一些镜像将被提升为leader。 如果继续停止节点，那么将到达一个镜像队列不再有镜像的点：它仅存在于一个节点上，且该节点上它是leader。 如果它的最后一个剩余节点关闭，但是镜像队列被声明为持久的，那么队列中的持久消息将在该节点重新启动后继续存在。

然而，镜像目前无法知道它的队列内容是否已经偏离了它重新加入的leader。 因此，当一个镜像重新加入一个镜像队列时， 它会丢弃已经拥有的任何持久本地内容并开始为空 。

默认情况下，RabbitMQ 将拒绝leader节点在受控关闭（即明确停止 RabbitMQ 服务或关闭操作系统）时提升非同步镜像，以避免消息丢失； 相反，整个队列将关闭，就好像未同步的镜像不存在一样。

leader节点不受控制的关闭（即服务器或节点崩溃，或网络中断）仍将触发未同步镜像的提升。

如果您希望在所有情况下都让leader队列移动到未同步的镜像（即，您会选择队列的可用性而不是避免由于未同步的镜像升级而导致的消息丢失），那么将 ha-promote-on-shutdown 策略键设置为 always 而不是比它的默认值 when-synced 。

如果 ha-promote-on-failure 策略键设置为 when-synced ，则即使 ha-promote-on-shutdown 键设置为 always ，也不会提升未同步的镜像。 这意味着如果leader节点发生故障，队列将变得不可用，直到leader恢复。 如果leader队列永久丢失，队列将不可用，除非它被删除（这也将删除其所有内容）并重新声明。

当队列的所有镜像都关闭时，可能会丢失队列的leader。 在正常操作中，队列关闭的最后一个节点将成为leader，该节点在再次启动时仍然是leader（因为它可能收到了其他镜像没有看到的消息）。

但是，当您调用 rabbitmqctl forget_cluster_node 时，RabbitMQ 将尝试为每个在我们忘记的节点上有其领导者的队列找到当前停止的镜像，并在它再次启动时“提升”该镜像成为新的领导者。 如果有多个候选者，将选择最近停止的镜像。

重要的是要理解 RabbitMQ 只能在 forget_cluster_node 期间提升停止的镜像，因为任何再次启动的镜像都会清除它们的内容，如上面“停止节点和同步”中所述。 因此在停止的集群中移除丢失的leader时，您必须在再次启动镜像之前调用 rabbitmqctl forget_cluster_node 。

消息中间件——RabbitMQ（四）命令行与管控台的基本操作！

在前面的文章中我们介绍过RabbitMQ的搭建： RabbitMQ的安装过 以及各大主流消息中间件的对比： ，本章就主要来介绍下我们之前安装的管控台是如何使用以及如何通过命令行进行操作。

rabbitmqctl stop_app：关闭应用

rabbitmqctl start_app：启动应用

rabbtmqctl status：节点状态

rabbitmqctl add_user username password：添加用户

rabbitmqctl list_users：列出所有用户

rabbitmqctl delete_user username：删除用户

rabbitmqctl clear_permissions - p vhostpath username: 清除用户权限

rabbitmqctl list_user_permissions_username: 列出用户权限

rabbitmqctl change_password username newpassword：修改密码

rabbitmqctl set_permissions -p vhostpath username ". " ". " ".*" ：设置用户权限

涉及的用户命令还有许多，这里就不一一列举了。

rabbitmqctl add_vhost vhostpath:创建虚拟主机

rabbitmqctl list_vhosts:列出所有虚拟主机

rabbitmqctl list_permissions -p vhostpath:列出虚拟主机上所有权限

rabbitmqctl delete_vhost vhostpath:删除虚拟主机

rabbitmqctl list_queues：查看所有队列信息

rabbitmqctl -p vhostpath purge_queue bule：清除队列里的消息

rabbitmqctl reset：移除所有数据，要在rabbitmqctl stop_app之后使用

rabbitmqctl join_clust clusternode [--ram]:组成集群命令

rabbitmqctl clustr_status:查看集群状态

rabbitmqctl change_cluster_node_type disc|ram 修改集群节点的存储形式

rabbitmqctl forget_cluster_node [--offline] 忘记节点（摘除节点）

rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2..] (修改节点名称)

登录RabbitMQ web管理界面：

默认的初始用户名和密码为： guest

从图中可以看到:

Broker的属性

定义由 用户 ， 虚拟主机 ， 权限 ， 参数 ， 交换 ， 队列 和 绑定 组成。 它们不包括队列的内容或集群名称。 独占队列不会被导出。

导入的定义将与当前定义合并。 如果在导入过程中发生错误，则所做的任何更改都不会回滚。

连接的属性

通道的属性

交换器属性

在Admin界面可以添加用户或者虚拟主机等操作。

参考文章：

;wfr=spiderfor=pc

推荐文章：

消息中间件——RabbitMQ（一）Windws/Linux环境搭建（完整版）

消息中间件——RabbitMQ（二）各大主流消息中间件综合对比介绍！

消息中间件——RabbitMQ（三）理解RabbitMQ核心概念和AMQP协议！