RabbitMQ部署手册

Docker部署

创建自签名证书

生成CA私钥和CA证书

[root@ecs ~]# mkdir -p /data/rabbitmq/ssl && cd /data/rabbitmq/ssl
[root@ecs ssl]# openssl req -x509 -newkey rsa:4096 -days 3650 -nodes \
-keyout ca.key -out ca.crt \
-subj "/C=CN/ST=Jiangsu/L=Nanjing/O=Maisi/OU=R&D/CN=RabbitMQ"

生成服务端私钥和证书请求

[root@ecs ssl]# openssl req -newkey rsa:4096 -nodes \
-keyout server.key -out server.csr \
-subj "/C=CN/ST=Jiangsu/L=Nanjing/O=Maisi/OU=R&D/CN=RabbitMQ"

创建SAN（Subject Alternative Name，主题备用名称）扩展文件

[root@ecs ssl]# cat > v3.ext <<EOF
subjectAltName = @alt_names

[alt_names]
DNS.1 = rabbitmq.local
DNS.2 = localhost
IP.1 = 192.168.1.3
IP.2 = 127.0.0.1
EOF

注：需要将192.168.1.3替换RabbitMQ真实的公网/内网地址。

使用CA签发服务端证书

[root@ecs ssl]# openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key \
-CAcreateserial -out server.crt -days 3650 -extfile v3.ext

运行rabbitmq容器

[root@ecs ~]# cat > /data/rabbitmq/rabbitmq.conf <<EOF
default_vhost = /

listeners.ssl.default = 5671
ssl_options.cacertfile = /etc/rabbitmq/ssl/ca.crt
ssl_options.certfile = /etc/rabbitmq/ssl/server.crt
ssl_options.keyfile = /etc/rabbitmq/ssl/server.key
ssl_options.verify = verify_peer
ssl_options.fail_if_no_peer_cert = false

management.tcp.port = 15672

management.ssl.port = 15671
management.ssl.cacertfile = /etc/rabbitmq/ssl/ca.crt
management.ssl.certfile = /etc/rabbitmq/ssl/server.crt
management.ssl.keyfile = /etc/rabbitmq/ssl/server.key
EOF
[root@ecs ~]# docker run --name=rabbitmq \
-p 5671-5672:5671-5672 \
-p 15671-15672:15671-15672 \
-e RABBITMQ_DEFAULT_USER=root \
-e RABBITMQ_DEFAULT_PASS=root \
--hostname rabbitmq \
-v /data/rabbitmq:/var/lib/rabbitmq \
-v /data/rabbitmq/ssl:/etc/rabbitmq/ssl/ \
-v /data/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro \
-d registry.cn-shanghai.aliyuncs.com/maisi/rabbitmq:4.2.2

--hostname是为了固定RabbitMQ的节点名称，从而保证数据的持久化。不指定--hostname，Docker创建容器时，会以容器ID作为主机名。RabbitMQ的数据（队列、交换机、用户、Vhost）保存在/var/lib/rabbitmq/mnesia/rabbit@[主机名]下，即使-v /data/rabbitmq:/var/lib/rabbitmq也会因为机器名不一致而无法直接使用之前的容器数据。

不指定RABBITMQ_DEFAULT_USER，rabbitmq默认会创建guest用户，并设置密码为guest。

常用端口说明：

端口	协议	说明
5672	AMQP	客户端连接端口应用程序（java、node等）与RabbitMQ进行消息收发通信的端口
5671	AMQP（TLS/SSL）	客户端加密传输连接端口
15672	HTTP	Web管理端口
15671	HTTP（TLS/SSL）	Web加密传输管理端口
1883	MQTT	客户端连接端口
15675	MQTT	客户端WebSocket传输连接端口
4369	EPMD Erlang Port Mapper Daemon	Erlang节点发现端口
25672	Erlang Distribution	节点间通信与CLI端口
15692	Prometheus Metrics	Prometheus监控端口

防火墙开放端口

[root@ecs ~]# firewall-cmd --zone=public --add-port=5671-5672/tcp --permanent
[root@ecs ~]# firewall-cmd --zone=public --add-port=15671-15672/tcp --permanent
[root@ecs ~]# firewall-cmd --zone=public --add-port={1883,15675}/tcp --permanent
[root@ecs ~]# firewall-cmd --reload

命令行操作

插件管理

开启Web管理

[root@ecs ~]# docker exec -it rabbitmq /bin/bash
root@20104fa324e1:/# rabbitmq-plugins enable rabbitmq_management

查看RabbitMQ监听的所有端口：

root@20104fa324e1:/# rabbitmq-diagnostics listeners
  Asking node rabbit@rabbitmq to report its protocol listeners ...
  Interface: [::], port: 15692, protocol: http/prometheus, purpose: Prometheus exporter API over HTTP
  Interface: [::], port: 25672, protocol: clustering, purpose: inter-node and CLI tool communication
  Interface: [::], port: 5672, protocol: amqp, purpose: AMQP 0-9-1 and AMQP 1.0
  Interface: [::], port: 5671, protocol: amqp/ssl, purpose: AMQP 0-9-1 and AMQP 1.0 over TLS
  Interface: [::], port: 15671, protocol: https, purpose: HTTP API over TLS (HTTPS)

Web管理地址：

开启延时消息插件

rabbitmq 4.2.2

[root@ecs ~]# cd /usr/local/src
[root@ecs src]# wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v4.2.0/rabbitmq_delayed_message_exchange-4.2.0.ez
[root@ecs src]# docker cp ./rabbitmq_delayed_message_exchange-4.2.0.ez rabbitmq:/plugins
                                         Successfully copied 44.5kB to rabbitmq:/opt/rabbitmq/plugins
[root@ecs src]# docker exec -it rabbitmq /bin/bash
root@20104fa324e1:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@20104fa324e1:
rabbitmq_delayed_message_exchange
The following plugins have been configured:
  rabbitmq_delayed_message_exchange
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_prometheus
  rabbitmq_web_dispatch
Applying plugin configuration to rabbit@20104fa324e1...
The following plugins have been enabled:
  rabbitmq_delayed_message_exchange

started 1 plugins.

rabbitmq 4.0.4

[root@ecs ~]# cd /usr/local/src
[root@ecs src]# wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v4.0.7/rabbitmq_delayed_message_exchange-v4.0.7.ez

注：

rabbitmq_delayed_message_exchange和rabbitmq的主次版本号（x.y.z的x.y）必须一致，否则会无法兼容。例如：rabbitmq_delayed_message_exchange-4.2.0.ez可以用于rabbitmq:4.2.2，但不用用于rabbitmq:4.0.4。
开启延时消息插件后，必须重启rabbitmq容器才能生效。

开启MQTT插件
```
root@20104fa324e1:/# rabbitmq-plugins enable rabbitmq_web_mqtt
```
开启rabbitmq_web_mqtt会自动开启rabbitmq_mqtt。mqtt连接参数示例值：
ws mqtt
Name mqtt mqtt
Host ws://192.168.1.54 mqtt://192.168.1.54
Port 15675 1883
Path /ws
Username root root
Password dcIvMA3EHignTxAV dcIvMA3EHignTxAV

	ws	mqtt
Name	mqtt	mqtt
Host	ws://192.168.1.54	mqtt://192.168.1.54
Port	15675	1883
Path	/ws
Username	root	root
Password	dcIvMA3EHignTxAV	dcIvMA3EHignTxAV

用户管理

添加用户

root@20104fa324e1:/# rabbitmqctl add_user maisi mM1lO8nN2sO4eK8q

修改用户密码

root@20104fa324e1:/# rabbitmqctl change_password maisi mM1lO8nN2sO4eK8q

设置用户角色
```
root@20104fa324e1:/# rabbitmqctl set_user_tags maisi administrator
```
设置administrator后即可登录Web Management UI。

删除指定用户

root@20104fa324e1:/# rabbitmqctl delete_user maisi

查看所有用户

root@20104fa324e1:/# rabbitmqctl list_users

vhost管理

添加vhost
```
root@20104fa324e1:/# rabbitmqctl add_vhost maisi
```
注：maisi和/maisi是2个不同的vhost，vhost名称允许有/路径符号。
设置用户权限
用户maisi在虚拟主机maisi下拥有Configure、Write和Read任意资源的权限。
```
root@20104fa324e1:/# rabbitmqctl set_permissions -p maisi maisi ".*" ".*" ".*"
```
- -p：指定vhost。
- ".*" ".*" ".*"：.*是正则表达式，表示任意字符。分别对应Configure（配置任意名称的Exchange和Queue）、Write（给任意名称的Exchange或Queue发送消息）和Read（从任意名称的Queue读取消息）。

查看所有vhost

root@20104fa324e1:/# rabbitmqctl list_vhosts

删除指定vhost

root@20104fa324e1:/# rabbitmqctl delete_vhost maisi

队列管理

查看队列列表

root@20104fa324e1:/# rabbitmqctl list_queues -p maisi

删除指定队列

root@20104fa324e1:/# rabbitmqctl delete_queue default-delay -p maisi

客户端调用

Node.JS

const amqp = require('amqplib');
const fs = require('fs');
const path = require('path');

(async () => {
    const username = 'maisi';
    const password = 'mM1lO8nN2sO4eK8q';
    const host = '192.168.1.3';
    const port = 5672;
    const vhost = 'maisi';

    try {
        /*
        const connection = await amqp.connect(`amqp://${username}:${password}@${host}:${port}/${vhost}?heartbeat=60&frameMax=8192`);
        */

        const connection = await amqp.connect({
            protocol: 'amqps',
            hostname: host,
            port: 5671,
            username,
            password,
            vhost,
            heartbeat: 60,
            frameMax: 8192
        }, {
            //报错：Error: self-signed certificate in certificate chain
            //方法一：忽略自签名证书校验
            rejectUnauthorized: false,
            //方法二：添加CA
            rejectUnauthorized: true,
            ca: [fs.readFileSync(path.join(__dirname, 'ca.crt'))]
        });
        console.log({ connection });
    } catch (e) {
        console.log(e);
    }
})();

注：RabbitMQ从4.1.0开始，RabbitMQ官方为了提高性能和安全性，将AMQP协议中frame_max（最大帧大小）的最小允许值从4096字节提升到了8192字节。如果amqplib版本较旧，连接RabbitMQ时最大帧大小默认使用的是4096，不符合新版RabbitMQ的最低要求，RabbitMQ就会拒绝连接，并抛出异常：failed to negotiate connection parameters: negotiated frame_max = 4096 is lower than the minimum allowed value (8192)。

解决方法（任选其一）：

升级amqplib到最新版（当前：2.0.1）；

修改连接参数，添加frameMax=8192：

const url = `amqp://${username}:${password}@${hostname}:${port}/${vhost}?heartbeat=60&frameMax=8192`;

Python

import pika

host = '192.168.1.3'
port = 5672
username = 'maisi'
password = 'mM1lO8nN2sO4eK8q'
vhost = 'maisi'

try:
    credentials = pika.PlainCredentials(username, password)
    # connection = pika.BlockingConnection(pika.ConnectionParameters(host, port, vhost, credentials))
    connection = pika.BlockingConnection(pika.ConnectionParameters(
        host=host,
        port=port,
        virtual_host=vhost,
        credentials=credentials
    ))
    channel = connection.channel()
    print(channel)
except Exception as e:
    print(e)
finally:
    if 'connection' in locals() and connection.is_open:
        connection.close()