RabbitMQ安装部署

Posted 2020-12-19 qq-460374192

1.1 RabbitMQ安装

1.1.1 安装erlang环境

1.1.1.1 安装GCC GCC-C++ Openssl等模块

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

1.1.1.2 安装ncurses

yum -y install ncurses-devel

1.1.1.3 下载erlang

切换到home路径下，下载erlang的安装包

cd /home wget http://erlang.org/download/otp_src_21.0.tar.gz

1.1.1.4 解压erlang并将解压后的文件改名为erlang

tar -xvf otp_src_21.0.tar.gz mv otp_src_21.0 ./erlang

1.1.1.5 使用./configure，检查编译环境并配置安装路径，将目录切换到erlang，运行如下命令：

cd erlang/ ./configure --prefix=/usr/local/erlang --without-javac

1.1.1.6 使用make 对源代码进行编译（在erlang目录下）

make

1.1.1.7 使用make install 开始安装（在erlang目录下）

make install

1.1.1.8 配置erlang环境变量 vi /etc/profile

vi /etc/profile #set erlang enviorment export PATH=$PATH:/user/local/erlang/bin #为了使环境变量生效，执行 source ~/.bashrc

1.1.1.9 创建软链：执行命令

ln -s /usr/local/erlang/bin/erl /usr/local/bin/erl

 

1.1.1.10 验证

erl

输入erl 后会进入erlang的命令行模式

 

1.1.2 安装rabbitmq-server

1.1.2.1 下载rabbitmq-server-3.7.8.tar.xz（在home下）

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.8/rabbitmq-server-generic-unix-3.7.8.tar.xz

1.1.2.2 由于下载的文件是xz文件，对于下载xz包进行解压，首先先下载xz压缩工具

yum install xz

1.1.2.3 对rabbitmq包进行解压：

xz -d rabbitmq-server-generic-unix-3.7.8.tar.xz tar -xvf rabbitmq-server-generic-unix-3.7.8.tar

1.1.2.4 随后移动至/usr/local/下 改名rabbitmq：

cp -r rabbitmq_server-3.7.8 /usr/local/rabbitmq

 

1.1.2.5 配置rabbitmq的环境变量

这种下载的方式解压后直接可以使用，无需再编译安装；
进入到rabbit文件内，其命令文件存在于sbin文件夹下，因此需要将sbin文件夹的路径添加到PATH中：修改/etc/profile [ vi /etc/profile ]

vi /etc/profile #set rabbitmq enviorment export PATH=/usr/local/rabbitmq/sbin:$PATH #为了使环境变量生效，执行 source ~/.bashrc

 

1.1.2.6 启用MQ管理方式：

rabbitmq-plugins enable rabbitmq_management   #启动后台管理

 

rabbitmq-server & #运行rabbitmq（&为后台执行）

 

 

1.1.2.7 开放端口，循序外部访问

firewall-cmd --add-port=5672/tcp firewall-cmd --add-port=15672/tcp firewall-cmd --add-port=25672/tcp firewall-cmd --add-port=4369/tcp

1.1.2.8 添加用户和权限

#1.添加用户 rabbitmqctl add_user admin admin #2.添加权限 rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*" #3.修改用户角色 rabbitmqctl set_user_tags admin administrator

1.1.2.9 验证

访问http://IP:15672/#/，用刚才创建的用户进行登录。

 

1.1.3 配置集群

1.1.3.1 环境准备

准备两台机器，均安装rabbitmq-server，分别配置ip为192.168.187.180，192.168.187.208

1.1.3.2 修改hosts文件，添加如下内容（iphostname）

192.168.187.180 192-168-187-180

192.168.187.208 192-168-187-208

1.1.3.3 复制.erlang.cookie文件

1.说明：

rabbitmq的集群是依附于erlang的集群来工作的，所以必须先构建起erlang的集群镜像。
(.erlang.cookie是erlang分布式的token文件，集群内所有的设备要持有相同的.erlang.cookie文件才允许彼此通信)。

2.查找.erlang.cookie文件

find / -name *.cookie

 

3.将查找到的.erlang.cookie文件，分别复制到另外两台虚拟机中。（过程中要输入另外两台虚拟机的密码）

scp /var/lib/rabbitmq/.erlang.cookie 192.168.187.180:/var/lib/rabbitmq/ scp /root/.erlang.cookie 192.168.187.180:/root/

4.复制完成后，需要修改文件权限

chown rabbitmq:rabbitmq /root/.erlang.cookie chmod 400 /root/.erlang.cookie chown rabbitmq:rabbitmq /var/lib/rabbitmq/.erlang.cookie chmod 400 /var/lib/rabbitmq/.erlang.cookie

1.1.3.4 .erlang.cookie复制完成后,逐个重启节点服务

systemctl restart rabbitmq-server.service

1.1.3.5 将192.168.187.208上的rabbitmq作为集群主节点，在节点rabbit@192-168-187-180上面执行如下命令，以加入集群中

rabbitmqctl stop_app rabbitmqctl reset rabbitmqctl join_cluster --ram rabbit@192-168-187-208 rabbitmqctl start_app

1.1.3.6 查看集群状态

rabbitmqctl cluster_status

1.1.3.7 访问web管理界面

 

1.1.4 四种集群架构

1.1.4.1 普通集群

集群中有两种节点：
1、 内存节点：只保存状态到内存（一个例外的情况是：持久的queue的持久内容将被保存到disk）
2、 磁盘节点：保存状态到内存和磁盘。
内存节点虽然不写入磁盘，但是它执行比磁盘节点要好。集群中，只需要一个磁盘节点来保存状态 就足够了如果集群中只有内存节点，那么不能停止它们，否则所有的状态，消息等都会丢失。

良好的设计架构可以如下：在一个集群里，有3台以上机器，其中1台使用磁盘模式，其它使用内存模式。其它几台为内存模式的节点，无疑速度更快，因此客户端（consumer、producer）连接访问它们。而磁盘模式的节点，由于磁盘IO相对较慢，因此仅作数据备份使用。

 

1.1.4.2 主备模式

实现RabbitMQ的高可用集群，一般在并发和数据量不高的情况下，这种模式非常的好且简单。主备模式也称为Warren模式

主备模式:主节点提供读写，从节点不提供读写服务，只是负责提供备份服务,备份节点的主要功能是在主节点宕机时，完成自动切换 从-->主

主备模式：所谓rabbitmq另外一种模式就是warren（兔子窝），就是一个主/备方案（主节点如果挂了，从节点提供服务而已，和activemq利用zookeeper做主/备一样）

HaProxy配置：

listen rabbitmq_cluster bind 0.0.0.0:5672 mode tcp  #配置TCP模式 balance roundrobin #简单的轮询 server bhz76 192.168.11.12:5672 check inter 5000 rise 2 fall 3 #主节点 server bhz77 192.168.11.13:5672 backup check inter 5000 rise 2 fall 3 #备用节点

rabbitMQ主从之间的数据复制是异步的，但是在rabbitmq中不会出丢数据的情况，这是因为rabbitmq的接口也是异步的，主收到一条消息写入本地存储，然后在发起写入从的请求。当所有从写入成功后，主才会给client返回ack说这次写入成功了。

1.1.4.3 远程模式（不常用）

 

1.1.4.4 镜像模式

其实质和普通模式不同之处在于，消息实体会主动在镜像节点间同步，而不是在consumer取数据时临时拉取。该模式带来的副作用也很明显，除了降低系统性能外，如果镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通讯大大消耗掉。所以在对可靠性要求较高的场合中适用，一个队列想做成镜像队列，需要先设置policy，然后客户端创建队列的时候，rabbitmq集群根据“队列名称”自动设置是普通集群模式或镜像队列。具体如下：

队列通过策略来使能镜像。策略能在任何时刻改变，rabbitmq队列也近可能的将队列随着策略变化而变化；非镜像队列和镜像队列之间是有区别的，前者缺乏额外的镜像基础设施，没有任何slave，因此会运行得更快。

为了使队列称为镜像队列，你将会创建一个策略来匹配队列，设置策略有两个键“ha-mode和 ha-params（可选）”。ha-params根据ha-mode设置不同的值，下面表格说明这些key的选项。

 

语法讲解：

 

在cluster中任意节点启用策略，策略会自动同步到集群节点

 

rabbitmqctl set_policy-p/ha-all"^"‘{"ha-mode":"all"}‘

 

这行命令在vhost名称为hrsystem创建了一个策略，策略名称为ha-allqueue,策略模式为 all 即复制到所有节点，包含新增节点，策略正则表达式为 “^” 表示所有匹配所有队列名称。例如：

 

rabbitmqctl set_policy-p/ha-all"^message"‘{"ha-mode":"all"}‘

 

注意：“^message” 这个规则要根据自己修改，这个是指同步“message”开头的队列名称，我们配置时使用的应用于所有队列，所以表达式为“^”

1.1.4.5 多活模式

• 多活模式：这种模式也是实现异地数据复制的主流模式，因为远程模式配置比较复杂，所以一般来说实现异地集群都是使用双活或者多活模式来实现的。这种模式需要依赖rabbitmq的federation插件，可以实现继续的可靠AMQP数据通信，多活模式在实际配置与应用非常的简单。
• RabbitMQ部署架构采用双中心模式（多中心），那么在两套（或多套）数据中心中各部署一套RabbitMQ集群，各中心之间还需要实现部分队列消息共享。多活集群架构如下：
• Federation插件是一个不需要构建Cluster，而在Brokers之间传输消息的高性能插件，Federation插件可以在Brokers或者Cluster之间传输消息，连接双方可以使用不同的users和vistual hosts，双方也可以使用版本不同的RabbitMQ和Erlang。Federation插件使用AMQP协议通信，可以接收不连续的传输。

Federation Exchanges,可以看成Downstream从Upstream主动拉取消息，但并不是拉取所有消息，必须是在Downstream上已经明确定义Bindings关系的Exchange，也就是有实际的物理Queue来接收消息，才会从Upstream拉取消息到Downstream。使用AMQP协议实施代理间通信，Downstream会将绑定关系组合在一起，绑定/解绑命令将会发送到Upstream交换机。因此，FederationExchange只接收具有订阅的消息。

2 运行维护

// 根据运维工作需要，提炼有针对性的操作维护指令或配置

3 性能优化

// 针对性能方面，提炼有针对性的操作维护指令或配置方式，优化系统性能

 

4 常见问题FAQ

5 名词解释

5.1 、ConnectionFactory、Connection、Channel

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

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

ConnectionFactory为Connection的制造工厂。

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

5.2 、Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

 

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

 

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

 

5.3 、Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。

这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

5.4 、Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

5.5 、Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

 

5.6 、Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。

 

Exchange是按照什么逻辑将消息路由到Queue的？这个将在下面的8、Binding中介绍。

RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在下面的10、Exchange Types中介绍。

5.7 、routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。

在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的长度限制为255 bytes。

5.8 、Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

 

5.9 、Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。

在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。
binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

5.10 、Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

 

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。

 

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

topic

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

binding key与routing key一样也是句点号“. ”分隔的字符串。

binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）。

 

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

headers

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

5.11 、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属性分析哪条请求被执行了，根据执行结果进行后续业务处理。