也就是一个主/备方案，主节点提供读写，备用节点不提供读写。如果主节点挂了，就切换到备用节点，原来的备用节点升级为主节点提供读写服务，当原来的主节点恢复运行后，原来的主节点就变成备用节点，和 activeMQ 利用 zookeeper 做主/备一样，也可以一主多备。
HaProxy 配置:
listen rabbitmq_cluster
bind 0000:567 # 配置 tcp 模式
mode tcp # 简单的轮询
balance roundrobin # 主节点 roundrobin 随机
server 你的76机器 hostname 1921681176:5672 check inter 5000 rise 2 fall 2
server 你的77机器 hostname 1921681177:5672 backup check inter 5000 rise 2 fall 2 # 备用节点
注意了，上面的 rabbitMQ 集群节点配置 # inter 每隔 5 秒对 mq 集群做健康检查， 2 次正确证明服务可用，2 次失败证明服务器不可用，并且配置主备机制
远程模式可以实现双活的一种模式，简称 shovel 模式，所谓的 shovel 就是把消息进行不同数据中心的复制工作，可以跨地域的让两个 MQ 集群互联，远距离通信和复制。
Shovel 就是我们可以把消息进行数据中心的复制工作，我们可以跨地域的让两个 MQ 集群互联。
如图所示，有两个异地的 MQ 集群（可以是更多的集群），当用户在地区 1 这里下单了，系统发消息到 1 区的 MQ 服务器，发现 MQ 服务已超过设定的阈值，负载过高，这条消息就会被转到 地区 2 的 MQ 服务器上，由 2 区的去执行后面的业务逻辑，相当于分摊我们的服务压力。
在使用了 shovel 插件后，模型变成了近端同步确认，远端异步确认的方式，大大提高了订单确认速度，并且还能保证可靠性。
如上图所示，当我们的消息到达 exchange，它会判断当前的负载情况以及设定的阈值，如果负载不高就把消息放到我们正常的 warehouse_goleta 队列中，如果负载过高了，就会放到 backup_orders 队列中。backup_orders 队列通过 shovel 插件与另外的 MQ 集群进行同步数据，把消息发到第二个 MQ 集群上。
这是 rabbitMQ 比较早期的架构模型了，现在很少使用了。
shovel 集群的配置，首先启动 rabbitmq 插件，命令如下:
rabbitmq-plugins enable amqp_client
rabbitmq-plugins enable rabbitmq_shovel
在 /etc/rabbitmq/ 目录下创建 rabbitmqconfig 文件。注意，我们源服务器和目的地服务器都使用这个相同的配置文件。
具体配置如下
非常经典的 mirror 镜像模式，保证 100% 数据不丢失。在实际工作中也是用得最多的，并且实现非常的简单，一般互联网大厂都会构建这种镜像集群模式。
mirror 镜像队列，目的是为了保证 rabbitMQ 数据的高可靠性解决方案，主要就是实现数据的同步，一般来讲是 2 - 3 个节点实现数据同步。对于 100% 数据可靠性解决方案，一般是采用 3 个节点。
集群架构如下
如上图所示，用 KeepAlived 做了 HA-Proxy 的高可用，然后有 3 个节点的 MQ 服务，消息发送到主节点上，主节点通过 mirror 队列把数据同步到其他的 MQ 节点，这样来实现其高可靠。
也是实现异地数据复制的主流模式，因为 shovel 模式配置比较复杂，所以一般来说，实现异地集群的都是采用这种双活 或者 多活模型来实现的。这种模式需要依赖 rabbitMQ 的 federation 插件，可以实现持续的，可靠的 AMQP 数据通信，多活模式在实际配置与应用非常的简单。
rabbitMQ 部署架构采用双中心模式(多中心)，那么在两套(或多套)数据中心各部署一套 rabbitMQ 集群，各中心的rabbitMQ 服务除了需要为业务提供正常的消息服务外，中心之间还需要实现部分队列消息共享。
多活集群架构如下:
federation 插件是一个不需要构建 cluster ，而在 brokers 之间传输消息的高性能插件，federation 插件可以在 brokers 或者 cluster 之间传输消息，连接的双方可以使用不同的 users 和 virtual hosts，双方也可以使用不同版本的 rabbitMQ 和 erlang。federation 插件使用 AMQP 协议通信，可以接受不连续的传输。federation 不是建立在集群上的，而是建立在单个节点上的，如图上的 rabbit node 3 可以与绿色的 node1、node2、node3 中的任意一个利用 federation 插件进行数据同步。
如上图所示，federation exchanges 可以看成 downstream 从 upstream 主动拉取消息，但是并不是拉取所有消息，必须是在 downstream 上已经明确定义 Bingdings 关系的 exchange，也就是有实际的物理 queue 来接收消息，才会从 upstream 拉取消息到 downstream 。
它使用 AMQP 协议实现代理间通信，downstream 会将绑定关系组合在一起，绑定/解绑命令将发送到 upstream 交换机。因此，federation exchange 只接收具有订阅的消息。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”的消息，不会转发“dogpuppy”等等。 它是完全匹配、单传播的模式。
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是有限制的，不能是任意的。格式是以点号“”分割的字符表。比如：”stockusdnyse”，“nysevmw”， “quickorangerabbit”。你可以放任意的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 为 “quickorangerabbit”将会发送到Q1和Q2中。rounting_key 为”lazyorangerabbithujjddd”会被投递到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
channelExchangeDeclare(ExchangeName,"direct", durable:true, autoDelete:false, arguments:null);//声明消息队列，且为可持久的
（2）queue持久化，在声明时指定durable => true
channelQueueDeclare(QueueName, durable:true, exclusive:false, autoDelete:false, arguments:null);//声明消息队列，且为可持久的
（3）消息持久化，在投递时指定delivery_mode => 2(1是非持久化)。
channelbasicPublish("", queueName, MessagePropertiesPERSISTENT_TEXT_PLAIN, msggetBytes());
如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的；如果exchange和queue两者之间有一个持久化，一个非持久化,则不允许建立绑定。
注意：一旦创建了队列和交换机，就不能修改其标志了。例如创建了一个non-durable的队列，然后想把它改变成durable的，唯一的办法就是删除这个队列然后重新创建。
4、Fair dispath 公平分发
你可能也注意到了，分发机制不是那么优雅，默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。n是取余后的，它不管Consumer是否还有unacked Message，只是按照这个默认的机制进行分发。那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却毫无休息的机会，那么Rabbit是如何处理这种问题呢
通过basicqos方法设置prefetch_count=1，如下设置
channelbasic_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，那么系统就会使用”amqdirect”这个exchange。我们创建一个queue时，默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去。如下：
channelBasicPublish("","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 架构为 rabbitMQ + keepalived 镜像模式。
rabbitmq01 1921681101
rabbitmq01 1921681102
vip 1921681110 在rabbitmq02 上
现在 rabbitmq01 故障 ，起不来，队列数据同步rabbbit02 有问题。
解决方法是将rabbitmq01 这个故障节点从集群中剔除，然后重新加入。
在rabbitmq02 1921681102服务器上
剔除rabbitmq01节点
# rabbitmqctl cluster_status
# rabbitmqctl forget_cluster_node rabbit@192-168-1-101
# rabbitmqctl cluster_status
在rabbitmq01 1921681101服务器上
停止rabbitmq相关进程
# systemctl stop rabbitmq-server
# ps aux | grep rabbit | grep -v grep | awk ‘{print $2}’| xargs kill -9
移除rabbitmq相关数据文件
# mkdir /kingdee/rabbitmqBackup
# mv /var/lib/rabbitmq/ /kingdee/rabbitmqBackup/
重新启动rabbitmq
# systemctl start rabbitmq-server
# ps aux | grep rabbit
创建rabbitMQ用户
# rabbitmqctl add_user mquser rabbitMQ@123
注：此密码为安装时设置的rabbitMQ密码
# rabbitmqctl list_users
# rabbitmqctl set_user_tags mquser administrator
# rabbitmqctl set_permissions -p / mquser '' '' ''
# rabbitmqctl stop_app
# rabbitmqctl join_cluster rabbit@192-168-1-102
# rabbitmqctl cluster_status
# rabbitmqctl start_app
# systemctl start keepalived
在rabbitmq02 1921681102服务器上
添加策略同步策略
# rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all","ha-sync-mode":"automatic"}'
RabbitMQ Cluster群集安装配置
>支持。查询rabbitmq的范围信息显示，使用范围都是固定的，都是有限制的，rabbitmq支持在国产服务器。服务器，也称伺服器，是提供计算服务的设备，由于服务器需要响应服务请求，并进行处理，因此通常来说服务器应具备承担服务并且保障服务的能力。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）