1. 博客
  2. 文章详情

一直在用的 rabbitmq 搭建教程，分享给大家

rabbitmq  
发布于 Aug 15, 2024

本文介绍公司一直在用的rabbitmq集群安装部署过程，版本不算太新，但一直稳定运行（哈哈，可能没啥量，大家看看就好）。

简介

官网：https://www.rabbitmq.com/

RabbitMQ 是一个开源的遵循 AMQP(Advanced Message Queuing Protocol) 协议实现的基于 Erlang 语言编写，支持多种客户端（语言），用于在分布式系统中存储消息，转发消息，具有高可用高可扩展性，易用性等特征。

AMQP ：高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布 / 订阅）、可靠性、安全。

总之，就一句话：RabbitMQ 是一个消息队列中间件。

安装

rabbitmq依赖erlang环境，类似java依赖jvm一样。

资源准备

机器列表

  • 10.100.16.201，centos7
  • 10.100.16.202，centos7
  • 10.100.16.203，centos7

软件版本

  • rabbitmq，3.8.14
  • erlang，23.2.7

下载地址

因为我这里只实践了 centos 这里有 rpm 包

https://github.com/rabbitmq/rabbitmq-server/releases/tag/v3.8.14

https://github.com/rabbitmq/erlang-rpm/releases/tag/v23.2.7

安装步骤

安装基础软件包

# centos 扩展软件库，
yum -y install epel-release 
# 高效的数据传输库，rabbitmq 需要
yum -y install socat

下载 rpm 包至本地然后直接安装

  • erlang-23.2.7-1.el7.x86_64.rpm
  • rabbitmq-server-3.8.14-1.el7.noarch.rpm
# 下载至本地安装
# 创建一个临时目录存放 rpm 包，安装后可以删除
mkdir -p /data/rabbitmq
cd /data/rabbitmq
wget https://github.com/rabbitmq/erlang-rpm/releases/download/v23.2.7/erlang-23.2.7-1.el7.x86_64.rpm
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.14/rabbitmq-server-3.8.14-1.el7.noarch.rpm

yum install -y erlang-23.2.7-1.el7.x86_64.rpm
yum install -y rabbitmq-server-3.8.14-1.el7.noarch.rpm

在 centos 上 rabbitmq 采用 systemd 管理启停比较方便

# 启动
systemctl start rabbitmq-server
# 查看状态
systemctl status rabbitmq-server
# 停止
systemctl stop rabbitmq-server
# 开机自启
systemctl enable rabbitmq-server

启动 rabbitmq 开启 管理控制台 插件（三台机器均需开启此插件）

rabbitmq-plugins enable rabbitmq_managementrabbitmq

启动后需要开启特定的端口放行 或者 关闭防火墙（懒人必备，慎用）

systemctl stop firewalld
systemctl disable firewalld

集群配置

3 台机器需要组成一个集群需要共享同一份 erlang cookie, 内容随意，保证 3 台相同即可。

echo 'AZVOCZYZZBVFLBPTBXU' > /var/lib/rabbitmq/.erlang.cookie
chown -R rabbitmq:rabbitmq /var/lib/rabbitmq/.erlang.cookie
chmod 400 /var/lib/rabbitmq/.erlang.cookie

配置 3 台机器的 /etc/hosts, 默认启动会采用 hostname 作为节点标识，确保其能通过 hostname 相互访问

10.100.16.201 prod-rabbitmq-16-201
10.100.16.202 prod-rabbitmq-16-202
10.100.16.203 prod-rabbitmq-16-203

启动 3 台 rabbitmq-server，并分别在 202,203 上执行加入集群的指令

systemctl start rabbitmq-server
 
 
# 在 202 上操作
# 1.停止服务
rabbitmqctl stop_app
# 2.重置状态
rabbitmqctl reset
# 3.节点加入
rabbitmqctl join_cluster rabbit@prod-rabbitmq-16-201
# 4.启动服务
rabbitmqctl start_app

 
# 在 203 上操作


rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@prod-rabbitmq-16-201
rabbitmqctl start_app

然后，可以在 201 上确认集群的状态

rabbitmqctl cluster_status
 
 
# 输出内容参考
Cluster status of node rabbit@prod-rabbitmq-16-201 ...
Basics
# 集群名称
Cluster name: rabbit@prod-rabbitmq-16-201
# 磁盘节点
Disk Nodes
 
rabbit@prod-rabbitmq-16-201
rabbit@prod-rabbitmq-16-202
rabbit@prod-rabbitmq-16-203
# 运行中节点
Running Nodes
 
rabbit@prod-rabbitmq-16-201
rabbit@prod-rabbitmq-16-202
rabbit@prod-rabbitmq-16-203
# 版本信息
Versions
 
rabbit@prod-rabbitmq-16-201: RabbitMQ 3.8.14 on Erlang 23.2.7
rabbit@prod-rabbitmq-16-202: RabbitMQ 3.8.14 on Erlang 23.2.7
rabbit@prod-rabbitmq-16-203: RabbitMQ 3.8.14 on Erlang 23.2.7
# 维护状态
Maintenance status
 
Node: rabbit@prod-rabbitmq-16-201, status: not under maintenance # 未在维护中
Node: rabbit@prod-rabbitmq-16-202, status: not under maintenance # 未在维护中
Node: rabbit@prod-rabbitmq-16-203, status: not under maintenance # 未在维护中
# 告警
Alarms
 
(none)
# 网络分区
Network Partitions
 
(none)
# 监听端口
Listeners
 
Node: rabbit@prod-rabbitmq-16-201, interface: [::], port: 15672, protocol: http, purpose: HTTP API
Node: rabbit@prod-rabbitmq-16-201, interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit@prod-rabbitmq-16-201, interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
Node: rabbit@prod-rabbitmq-16-202, interface: [::], port: 15672, protocol: http, purpose: HTTP API
Node: rabbit@prod-rabbitmq-16-202, interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit@prod-rabbitmq-16-202, interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
Node: rabbit@prod-rabbitmq-16-203, interface: [::], port: 15672, protocol: http, purpose: HTTP API
Node: rabbit@prod-rabbitmq-16-203, interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
Node: rabbit@prod-rabbitmq-16-203, interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
# 特性
Feature flags
 
Flag: drop_unroutable_metric, state: disabled
Flag: empty_basic_get_metric, state: disabled
Flag: implicit_default_bindings, state: enabled
Flag: maintenance_mode_status, state: enabled
Flag: quorum_queue, state: enabled
Flag: user_limits, state: enabled
Flag: virtual_host_metadata, state: enabled

用户配置

此版本 rabbitmq 默认有一个 guest/guest 账号，但是只能 localhost 方式使用，因此还需要再创建一个远程用户供客户端或管理控制面板使用

# 添加用户
# rabbitmqctl add_user 用户名 密码
rabbitmqctl add_user admin adminpwd
 
# 设置用户角色,分配操作权限
# rabbitmqctl set_user_tags 用户名 角色
rabbitmqctl set_user_tags admin administrator
 
# 为用户添加资源权限(授予访问虚拟机根节点的所有权限)
# rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"
rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

rabbitmq 有 4 类角色：

  • administrator：可以登录控制台、查看所有信息、并对 rabbitmq 进行管理
  • monToring：监控者；登录控制台，查看所有信息
  • policymaker：策略制定者；登录控制台指定策略
  • managment：普通管理员；登录控制

创建完成后，访问服务器 ip:15672 进行登录，然后便可进入到后台

负载均衡

rabbitmq 集群中的 3 台机器是平行的，但客户端一般只需要连接其中的一台，为了更好的分摊压力，可以通过 负载均衡 软硬件对其进行压力分配。

软件的安装不再赘述，只罗列配置参考。

nginx+keepalived

机器分配

采用2台nginx负载均衡

  • 10.100.16.211, 5672/15672
    • vip: 10.100.16.210, 5672/15672
  • 10.100.16.212, 5672/15672
    • vip: 10.100.16.210, 5672/15672

nginx 配置

nginx 需支持 stream 模块，可以执行 nginx -V 查看是否有 --with-stream

nginx.conf 中 http 模块平级 新增 stream 模块

...省略
 
 
http {
    ...省略
}
 
 
stream {
    upstream rabbitmq_15672 {
        server 10.100.16.201:15672;
        server 10.100.16.202:15672;
        server 10.100.16.203:15672;
    }
    server {
        listen 15672;
        proxy_connect_timeout 5s;
        proxy_timeout 24h;
        proxy_pass rabbitmq_15672;
    }
 
    upstream rabbitmq_5672 {
        server 10.100.16.201:5672;
        server 10.100.16.202:5672;
        server 10.100.16.203:5672;
    }
    server {
        listen 5672;
        proxy_connect_timeout 5s;
        proxy_timeout 24h;
        proxy_pass rabbitmq_5672;
    }
}

keepalived 配置

211 机器配置

global_defs {
   router_id prod-internal-nginx-16-211
}
 
vrrp_instance VI_1 {
    state MASTER
    interface ens192
    virtual_router_id 210
    priority 150
    advert_int 1
    authentication {
        auth_type PASS
        auth_pass admin@20240105$
    }
    virtual_ipaddress {
        10.100.16.210/32 dev ens192 label ens192:0
    }
}

212 机器配置

global_defs {
   router_id prod-internal-nginx-16-212
}
 
vrrp_instance VI_1 {
    state BACKUP
    interface ens192
    virtual_router_id 210
    priority 150
    advert_int 1
    authentication {
        auth_type PASS
        auth_pass admin@20240105$
    }
    virtual_ipaddress {
        10.100.16.210/32 dev ens192 label ens192:0
    }
}

haproxy+keepalived

原理基本同 nginx, 使用 haproxy 代替 nginx 而已。

参考博文：

https://www.cnblogs.com/caoweixiong/p/14411785.html

验证测试

使用 python 快速验证，安装 python3, pip3, pika

yum install python34 python34-pip
pip3.4 install pika
 
 
# 可以将 python34 设置为默认的 python
cd /usr/bin
mv python python.old
mv pip pip.old
ln -s python3.4 python
ln -s pip3.4 pip

编写 producer.py 和 consumer.py

producer.py

# coding: utf-8
import json
import pika
import datetime
 
# 生成消息
def get_message():
    for i in range(10): # 生成10条消息
        message=json.dumps({'id': "10000%s" % i, "amount": 100 * i,"name":"tony","createtime":str(datetime.datetime.now())})
        producter(message)
 
# 消息生产者
def producter(message):
    # 获取与 rabbitmq 服务的连接，虚拟队列需要指定参数 virtual_host，如果是默认的可以不填（默认为/)，也可以自己创建一个
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.100.16.210', port=5672,credentials=pika.PlainCredentials('admin', 'quanshi')))
    # 创建一个 AMQP 信道（Channel）
    channel = connection.channel()
    # 声明消息队列 tester，消息将在这个队列传递，如不存在，则创建
    channel.queue_declare(queue='tester')
    # 向队列插入数值 routing_key 的队列名为 tester，body 就是放入的消息内容，exchange 指定消息在哪个队列传递，这里使用空字符串（默认的exchange）
    channel.basic_publish(exchange='', routing_key='tester', body=message)
    # 关闭连接
    connection.close()
 
if __name__=="__main__":
    get_message() #程序执行入口

执行 python producer.py 即可

consumer.py

# coding:utf-8
import pika
 
# 接收消息，直接输出
def write_file(message):
    #with open("msg.txt","a+") as f:
    #    f.write(message)
    print(message)
 
 
def consumer():# 消息消费者
    # 获取与 rabbitmq 服务的连接
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.100.16.210', port=5672,credentials=pika.PlainCredentials('admin', 'quanshi')))
    # 创建一个 AMQP 信道（Channel）
    channel = connection.channel()
    # 声明消息队列 tester,durable=False 表示不持久化
    channel.queue_declare(queue='tester', durable=False)
    # 定义一个回调函数来处理消息队列中的消息，这里是将消息写入文件，你也可以入库。
    def callback(ch, method, properties, body):
        ch.basic_ack(delivery_tag=method.delivery_tag) # 告诉生成者，消息处理完成
        write_file(body.decode())
    # 消费tester列表里的消息，收到就调用callback函数
    channel.basic_consume('tester', callback)
    # 开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
    channel.start_consuming()
 
if __name__=="__main__":
    consumer()

执行 python consumer.py 开始消费消息。

监控告警

监控告警。

目前先配置简单的进程监控和端口监控。

配置端口指标采集，监控 5672 即可。

配置进程采集，rabbitmq 使用 erlang 虚拟环境启动，进程名：beam.smp

卸载

如果不需要此软件，可以按照如下步骤清理。

# 停掉服务
systemctl disable rabbitmq-server
systemctl stop rabbitmq-server
# kill 掉残留的 erlang 进场
ps -ef |grep rabbit

# 删掉 rabbitmq
yum list|grep rabbitmq
yum -y remove rabbitmq-server.noarch
 
# 删掉 erlang
yum list | grep erlang
yum -y remove erlang.x86_64

# 删除相关文件
rm -rf /usr/lib64/erlang
rm -rf /var/lib/rabbitmq
rm -rf /usr/lib/rabbitmq
rm -rf /etc/rabbitmq/
rm -rf /var/log/rabbitmq

总结

今天带大家快速熟悉了一下 rabbitmq 的搭建，内容比较简单，但是比较实用，希望对大家有所帮助。

本次的分享到此结束，希望对你有所帮助。

如果你对我分享的内容感兴趣，欢迎扫码关注公众号：新质程序猿，并设置星标，推送更实时哟！

本文由 黄彦祥 创作，采用 知识共享署名 3.0 中国大陆许可协议 进行许可。
可自由转载、引用，但需署名作者且注明文章出处。