淘先锋技术网

首页 1 2 3 4 5 6 7

Rabbitmq 消息队列

概念

队列
MQ 全称为 Message Queue, 消息队列（ MQ ）

是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。

消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

AMQP 即 Advanced Message Queuing Protocol

高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布 / 订阅）、可靠性、安全。

Rabbitmq

  • 属于一个流行的开源消息队列系统。属于AMQP( 高级消息队列协议 ) 标准的一个
    实现。是应用层协议的一个开放标准，为面向消息的中间件设计。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

  • 消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

  • AMQP的主要特征是面向消息、队列、路由（包括点对点和发布 / 订阅）、可靠性、安全RabbitMQ特点

  • 使用Erlang编写

  • 支持持久化

  • 支持HA

  • 提供C# , erlang，java，perl，python，ruby等的client开发端

Rabbitmq中概念名词

Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字， exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个
channel代表一个会话任务。

Rabbitmq 的 metadata

元数据可以持久化在 RAM 或 Disc. 从这个角度可以把 RabbitMQ 集群中的节点分成两种 :RAM Node和 Disk Node. RAM Node 只会将元数据存放在RAM Disk node 会将元数据久化到磁盘。

单节点系统就没有什么选择了 , 只允许 disk node, 否则由于没有数据冗余一旦重启就会丢掉所有的配置信息 . 但在集群环境中可以选择哪些节点是 RAM node. 在集群中声明 (declare) 创建exchange queue binding, 这类操作要等到所有的节点都完成创建才会返回 :
如果是内存节点就要修改内存数据 ,

如果是 disk node 就要等待写磁盘 , 节点过多这里的速度就会被大大的拖慢.

有些场景 exchang queue 相当固定 , 变动很少 ,那即使全都是 disc node, 也没有什么影响 . 如果使用 Rabbitmq 做 RPC（ RPC :Remote Procedure Call—远程过程调用）, RPC 或者类似 RPC 的场景这个问题就严重了 , 频繁创建销毁临时队列 , 磁盘读写能力就很快成为性能瓶颈了。所以 , 大多数情况下 , 我们尽量把 Node 创建为RAM Node. 这里就有一个问题了 , 要想集群重启后元数据可以恢复就需要把集群元数据持久化到磁盘 , 那需要规划 RabbitMQ 集群中的 RAM Node 和 Disc Node 。

只要有一个节点是 Disc Node 就能提供条件把集群元数据写到磁盘 ,RabbitMQ 的确也是这样要求的 : 集群中只要有一个 disk node 就可以 , 其它的都可以是 RAMnode. 节点加入或退出集群一定至少要通知集群中的一个 disk node 。

如果集群中 disk node 都宕掉 , 就不要变动集群的元数据 . 声明 exchange queue修改用户权限 , 添加用户等等这些变动在节点重启之后无法恢复 。

有一种情况要求所有的 disk node 都要在线情况在才能操作 , 那就是增加或者移除节点 .RAM node 动的时候会连接到预设的 disk node 下载最新的集群元数据 . 如果你有两个 disk node(d1 d2), 一RAM node 加入的时候你只告诉 d1, 而恰好这个RAM node 重启的时候 d1 并没有启动 , 重启就会败 . 所以加入 RAM 节点的时候 ,把所有的disk node 信息都告诉它 ,RAM node 会把 disk node 的信持久化到磁盘以便后续启动可以按图索骥 .

下载rabbitmq

环境：
chen1：192.168.1.13
chen2：192.168.1.14
chen2：192.168.1.15

配置dns全局解释器：
vim /etc/resolv.conf
nameserver 192.168.254.251 #连接learn.yunwei.edu的内网
nameserver 223.5.5.5

wget http://download2.yunwei.edu/shell/yum-repo.sh
sh yum-repo.sh

安装rabbtimq和erlang软件包
yum install -y erlang rabbitmq-server.noarch
systemctl enable rabbitmq-server.service
systemctl start rabbitmq-server.service

查看监听端口
netstat -lantp | grep 5672

以上步骤三台都需进行
配置文件
vim /etc/rabbitmq/rabbitmq.config

chen1:修改guest密码（默认用户为：guest，密码为guest）
rabbitmqctl change_password guest admin

chen1:编辑rabbittmq变量文件
vim /etc/rabbitmq/rabbitmq-env.conf

RABBITMQ_NODE_PORT=5672
ulimit -S -n 4096
RABBITMQ_SERVER_ERL_ARGS="+K true +A30 +P 1048576 -kernel inet_default_connect_options [{nodelay,true},{raw,6,18,<<5000:64/native>>}] -kernel inet_default_listen_options [{raw,6,18,<<5000:64/native>>}]"
RABBITMQ_NODE_IP_ADDRESS=192.168.1.13

编辑好之后发送至其它两个节点
scp /etc/rabbitmq/rabbitmq-env.conf chen2:/etc/rabbitmq/
scp /etc/rabbitmq/rabbitmq-env.conf chen3:/etc/rabbitmq/

chen2:

vim /etc/rabbitmq/rabbitmq-env.conf
RABBITMQ_NODE_PORT=5672
ulimit -S -n 4096
RABBITMQ_SERVER_ERL_ARGS="+K true +A30 +P 1048576 -kernel inet_default_connect_options [{nodelay,true},{raw,6,18,<<5000:64/native>>}] -kernel inet_default_listen_options [{raw,6,18,<<5000:64/native>>}]"
RABBITMQ_NODE_IP_ADDRESS=192.168.1.14

chen3:
vim /etc/rabbitmq/rabbitmq-env.conf

RABBITMQ_NODE_PORT=5672
ulimit -S -n 4096
RABBITMQ_SERVER_ERL_ARGS="+K true +A30 +P 1048576 -kernel inet_default_connect_options [{nodelay,true},{raw,6,18,<<5000:64/native>>}] -kernel inet_default_listen_options [{raw,6,18,<<5000:64/native>>}]"
RABBITMQ_NODE_IP_ADDRESS=192.168.1.15

所有节点开启rabbitmq的web管理页面
rabbitmq-plugins enable rabbitmq_management
在这里插入图片描述
账号：guest
密码：admin

chen1发送selang.cookie到其它节点配置集群 #这一步相当于指定谁是主，谁是从

scp /var/lib/rabbitmq/.erlang.cookie chen2:/var/lib/rabbitmq/.erlang.cookie
scp /var/lib/rabbitmq/.erlang.cookie chen3:/var/lib/rabbitmq/.erlang.cookie

chen2和chen3停止应用，并以ram的方式加入chen1节点，之后重启应用
这里的应有指的是rabbitmq服务给我们提供消息队列这个应用，而不是rabbitmq服务
systemctl restart rabbitmq-server.service #因为发送新的配置文件，得重启服务
查看状态是否启用
systemctl restart rabbitmq-server.service
在这里插入图片描述

停止应用
rabbitmqctl stop_app

rabbitmqctl stop_app
[root@chen3 ~]# rabbitmqctl stop_app
Stopping node rabbit@chen3 ...

chen2、chen3加入集群
rabbitmqctl join_cluster --ram rabbit@chen1 #加入集群，以ram形式的节点加入

重启应用
rabbitmqctl start_app

chen1检查集群状态
rabbitmqctl cluster_status

[root@chen1 ~]# rabbitmqctl cluster_status
Cluster status of node rabbit@chen1 ...
[{nodes,[{disc,[rabbit@chen1]},{ram,[rabbit@chen3,rabbit@chen2]}]},
 {running_nodes,[rabbit@chen2,rabbit@chen3,rabbit@chen1]},
 {cluster_name,<<"rabbit@chen1">>},
 {partitions,[]},
 {alarms,[{rabbit@chen2,[]},{rabbit@chen3,[]},{rabbit@chen1,[]}]}]

登录网站验证
http://192.168.1.13:15672/#/
在这里插入图片描述

由于其它节点已经加入进去，则密码就与chen1一致，为admin

其他命令：
（1）添加管理员：
rabbitmqctl add_user mqadmin mqadmin
rabbitmqctl set_user_tags mqadmin administrator
rabbitmqctl set_permissions -p / mqadmin “." ".” “.*”

（2）更改节点类型（内存型或磁盘型）
rabbitmqctl stop_app
rabbitmqctl change_cluster_node_type disc 或 rabbitmqctl change_cluster_node_type ram
rabbitmqctl start_app

（3）从集群移除节点（或者重置节点）
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl start_app
rabbitmqctl cluster_status

（4）从某个节点移除集群中其他节点
rabbitmqctl forget_cluster_node rabbit@node3
rabbitmqctl reset
rabbitmqctl start_app
rabbitmqctl cluster_status

注意

  1. 保证集群中至少有一个磁盘类型的节点以防数据丢失，在更改节点类型时尤其要注意。
  2. 若整个集群被停掉了，应保证最后一个 down 掉的节点被最先启动，若不能则要使用 forget_cluster_node 命令将其移出集群
  3. 若集群中节点几乎同时以不可控的方式 down 了此时在其中一个节点使用 force_boot 命令重启节点