技术的路上奔跑

入门

0%

rabbitmq学习笔记

发表于 更新于 分类于 ， Valine：
本文字数： 48k 阅读时长 ≈ 43 分钟

一、安装

1. Windows

1.1. 安装步骤

  • 下载Erlang OTP并安装
  • 下载RabbitMQ安装包（.exe）
  • 按照提示安装

1.2. 服务管理

1
2
3
4
5
6
7
8
:: 启动RabbitMQ服务
net start rabbitmq

:: 停止RabbitMQ服务
net stop rabbitmq

:: 重启RabbitMQ服务
net restart rabbitmq

1.3. 管理插件启用

1
2
3
4
5
:: 启用管理界面插件
rabbitmq-plugins enable rabbitmq_management

:: 禁用管理界面插件
rabbitmq-plugins disable rabbitmq_management

2. Linux (Ubuntu/Debian)

2.1. 安装Erlang

1
2
3
4
5
6
7
# 添加Erlang仓库
wget -O- https://packages.erlang-solutions.com/ubuntu/erlang_solutions.asc | sudo apt-key add -
echo "deb https://packages.erlang-solutions.com/ubuntu $(lsb_release -cs) contrib" | sudo tee /etc/apt/sources.list.d/erlang.list

# 更新并安装Erlang
sudo apt update
sudo apt install erlang

2.2. 安装RabbitMQ

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 添加RabbitMQ仓库
sudo apt install apt-transport-https
wget -O- https://www.rabbitmq.com/rabbitmq-release-signing-key.asc | sudo apt-key add -
echo 'deb https://dl.bintray.com/rabbitmq-erlang/debian $(lsb_release -cs) erlang' | sudo tee /etc/apt/sources.list.d/rabbitmq.list

# 更新并安装RabbitMQ
sudo apt update
sudo apt install rabbitmq-server

# 启动RabbitMQ服务
sudo systemctl start rabbitmq-server

# 设置开机自启
sudo systemctl enable rabbitmq-server

3. Docker安装

1
2
3
4
5
6
7
8
9
10
11
# 拉取RabbitMQ镜像
docker pull rabbitmq:management

# 运行容器
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

# 持久化数据
docker run -d --name rabbitmq \
  -p 5672:5672 -p 15672:15672 \
  -v rabbitmq_data:/var/lib/rabbitmq \
  rabbitmq:management

4. Kubernetes Helm安装

4.1. 前置条件

  • 已安装 Kubernetes 集群
  • 已安装 Helm 3
  • 已配置 kubectl 连接到集群
1
2
3
4
5
# 检查 Kubernetes 集群状态
kubectl cluster-info

# 检查 Helm 版本
helm version

4.2. 编写 Chart

1) 准备 Chart 文件

1
2
3
4
5
6
7
# 创建 Chart 目录结构
mkdir -p rabbitmq-chart
cd rabbitmq-chart

# 创建必要的文件
touch Chart.yaml values.yaml
mkdir -p templates

2) 配置 values.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
# 公共配置
replicas: 1
namespace: application

global:
  image:
    # 本地搭建的harbor仓库，镜像地址harbor-local.harbor/official/rabbitmq:4.1.0-management
    host: harbor-local.harbor
    branch: official
    version: 4.1.0-management

app:
  mode: standard        # 用于标识灰度标签，给istio使用
  gcPercent: 100

# 对应内部clusterIP的service
service:
  ports:
    - port: 5672
      name: tcp
      appProtocol: TCP
      protocol: TCP

# 对应nodePort的service
serviceExternal:
  ports:
    - port: 15672
      name: tcp
      appProtocol: TCP
      protocol: TCP
      # nodePort: 30180   # 可选，不设置就自动分配

# 资源限制配置
resources:
  requests:
    cpu: 200m
    memory: 1G
  limits:
    cpu: 4
    memory: 16G

# RabbitMQ 认证配置（base64加密）
rabbitmq:
  pass: cm9vdA==  # base64加密的 "root"
  user: cm9vdA==  # base64加密的 "root"

3) 创建模板文件

在 templates/ 目录下创建以下文件：

secret.yaml

1
2
3
4
5
6
7
8
9
apiVersion: v1
kind: Secret
metadata:
  name: {{ .Release.Name }}-secret
  namespace: {{ .Values.namespace }}
type: Opaque
data:
  user: {{ .Values.rabbitmq.user }}
  pass: {{ .Values.rabbitmq.pass }}

service.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
{{ if eq .Values.app.mode "standard" }}
apiVersion: v1
kind: Service
metadata:
  name: {{ .Chart.Name }}-svc
  namespace: {{ .Values.namespace }}
  annotations:
    helm.sh/resource-policy: "keep"
spec:
  selector:
    app: {{ .Chart.Name }}
  type: ClusterIP
  ports:
    {{- range .Values.service.ports }}
    - port: {{ .port }}
      targetPort: {{ .port }}
      name: {{.name}}
      appProtocol: {{.appProtocol}}
      protocol: {{.protocol}}
    {{- end}}
---
apiVersion: v1
kind: Service
metadata:
  name: {{ .Chart.Name }}-svc-external
  namespace: {{ .Values.namespace }}
  annotations:
    helm.sh/resource-policy: "keep"
spec:
  selector:
    app: {{ .Chart.Name }}
  type: NodePort
  ports:
    {{- range .Values.serviceExternal.ports }}
    - port: {{ .port }}
      targetPort: {{ .port }}
      name: {{.name}}
      appProtocol: {{.appProtocol}}
      protocol: {{.protocol}}
      {{- if index . "nodePort"}}
      nodePort: {{.nodePort}}
      {{- end }}
    {{- end}}
{{ end }}

statefulset.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ .Release.Name }}-{{ .Values.app.mode }}
  namespace: {{ .Values.namespace }}
  labels:
    app: {{ .Chart.Name }}
    version: {{ .Values.global.image.version }}
    appMode: {{ .Values.app.mode }}
spec:
  selector:
    matchLabels:
      app: {{ .Chart.Name }}
      appMode: {{.Values.app.mode }}
  replicas: {{ .Values.replicas }}
  template:
    metadata:
      labels:
        app: {{ .Chart.Name }}
        version: {{ .Values.global.image.version }}
        appMode: {{.Values.app.mode}}
        chartName: {{ .Chart.Name }}
        releaseName: {{ .Release.Name }}
    spec:
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 80
              podAffinityTerm:
                topologyKey: kubernetes.io/hostname
                labelSelector:
                  matchLabels:
                    releaseName: {{ .Release.Name }}
                    version: {{ .Values.global.image.version }}
      containers:
        - name: {{ .Chart.Name }}
          resources:
            requests:
              cpu: {{ .Values.resources.requests.cpu }}
              memory: {{ .Values.resources.requests.memory }}
            limits:
              cpu: {{ .Values.resources.limits.cpu }}
              memory: {{ .Values.resources.limits.memory }}
          image: {{ .Values.global.image.host }}/{{ .Values.global.image.branch }}/{{ .Chart.Name }}:{{.Values.global.image.version}}
          ports:
          {{- range .Values.service.ports }}
            - containerPort: {{ .port }}
              name: {{ .name }}
          {{- end}}
          env:
            - name: RABBITMQ_DEFAULT_USER
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-secret
                  key: user
            - name: RABBITMQ_DEFAULT_PASS
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-secret
                  key: pass
          volumeMounts:
          - mountPath: /etc/localtime
            name: tz-config
          readinessProbe:
            tcpSocket:
              {{- with index .Values.service.ports 0 }}
              port: {{ .port }}
              {{- end}}
            initialDelaySeconds: 10
            periodSeconds: 10
          livenessProbe:
            tcpSocket:
              {{- with index .Values.service.ports 0 }}
              port: {{ .port }}
              {{- end}}
            initialDelaySeconds: 10
            periodSeconds: 10
      volumes:
      - hostPath:
          path: /etc/localtime
          type: ""
        name: tz-config

4) 安装 Chart

1
2
3
4
5
6
7
8
9
10
11
# 创建命名空间
kubectl create namespace application

# 安装 Chart
helm install my-rabbitmq ./rabbitmq-chart -n application

# 查看安装状态
helm status my-rabbitmq -n application

# 查看 Pod 状态
kubectl get pods -n application -l app=rabbitmq

4.3. 访问 RabbitMQ

1
2
3
4
5
6
7
8
9
10
# 获取 NodePort
NODE_PORT=$(kubectl get --namespace application -o jsonpath="{.spec.ports[0].nodePort}" services my-rabbitmq-svc-external)
NODE_IP=$(kubectl get nodes --namespace application -o jsonpath="{.items[0].status.addresses[0].address}")

echo "RabbitMQ Management URL: http://$NODE_IP:$NODE_PORT"
echo "Username: "
echo "Password: admin123"

# 端口转发（本地访问）
kubectl port-forward --namespace application svc/my-rabbitmq-svc-external 15672:15672

4.4. 升级和卸载

1
2
3
4
5
6
7
8
# 升级 Release
helm upgrade my-rabbitmq ./rabbitmq-chart -n application

# 回滚到上一个版本
helm rollback my-rabbitmq -n application

# 卸载 Release
helm uninstall my-rabbitmq -n application

4.5. 配置持久化存储

1
2
3
4
5
6
7
8
# 在 values.yaml 中添加持久化配置
persistence:
  enabled: true
  storageClass: "standard"
  accessModes:
    - ReadWriteOnce
  size: 10Gi
  mountPath: /var/lib/rabbitmq

4.6. 配置资源限制

1
2
3
4
5
6
7
8
# 在 values.yaml 中配置资源限制
resources:
  requests:
    cpu: 200m
    memory: 512Mi
  limits:
    cpu: 1000m
    memory: 2Gi

4.7. 踩坑记

1) Erlang版本不兼容

1
2
3
4
5
# 查看Erlang版本
erl -version

# 查看RabbitMQ要求的Erlang版本
rabbitmqctl status

2) 端口冲突

  • 默认端口: 5672 (AMQP), 15672 (管理界面)
  • 检查端口占用
1
2
3
4
5
6
# Linux
netstat -tlnp | grep 5672
lsof -i :5672

# Windows
netstat -ano | findstr 5672

二、配置

1. 配置文件位置

  • Linux: /etc/rabbitmq/rabbitmq.conf
  • Windows: %APPDATA%\RabbitMQ\rabbitmq.conf

2. 基本配置示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 监听地址和端口
listeners.tcp.default = 5672

# 管理界面配置
management.tcp.port = 15672
management.tcp.ip = 0.0.0.0

# 内存使用限制
vm_memory_high_watermark.relative = 0.6

# 磁盘空间限制
disk_free_limit.relative = 1.0

# 日志配置
log.file.level = info
log.file = /var/log/rabbitmq/rabbit.log
log.console = true

# 集群配置
cluster_name = my_rabbitmq_cluster
cluster_keepalive_interval = 10000

# 默认用户和虚拟主机
default_user = guest
default_pass = guest
default_vhost = /

# 允许远程访问guest用户（仅开发环境）
loopback_users.guest = false

3. 配置查看

1
2
3
4
5
6
7
8
# 查看当前配置
rabbitmqctl environment

# 查看运行时参数
rabbitmqctl status

# 查看有效配置
rabbitmqctl eval 'application:get_all_env(rabbit).'

4. 配置说明

4.1. 内存管理

1
2
3
4
5
6
# 内存高水位标记，达到此值会阻塞生产者
vm_memory_high_watermark.relative = 0.6
vm_memory_high_watermark.absolute = 2GB

# 内存警报策略
memory_monitor_interval = 1000

4.2. 磁盘空间管理

1
2
3
# 磁盘空间限制，低于此值会阻塞生产者
disk_free_limit.relative = 1.0
disk_free_limit.absolute = 1GB

4.3. 集群配置

1
2
3
4
5
6
7
# 集群节点通信端口
cluster_nodes.disc = rabbit@node1,rabbit@node2
cluster_nodes.ram = rabbit@node3

# 集群同步配置
cluster_keepalive_interval = 10000
cluster_partition_handling = autoheal

三、命令操作

1. 基础管理命令

1.1. 服务管理

1
2
3
4
5
6
7
8
9
10
11
# 启动RabbitMQ
rabbitmq-server start

# 停止RabbitMQ
rabbitmq-server stop

# 重启RabbitMQ
rabbitmq-server restart

# 查看状态
rabbitmqctl status

1.2. 用户管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 添加用户
rabbitmqctl add_user username password

# 删除用户
rabbitmqctl delete_user username

# 修改密码
rabbitmqctl change_password username newpassword

# 设置用户标签
rabbitmqctl set_user_tags username administrator

# 查看用户列表
rabbitmqctl list_users

# 设置用户权限
rabbitmqctl set_permissions -p / username ".*" ".*" ".*"

# 查看用户权限
rabbitmqctl list_user_permissions username

1.3. 虚拟主机管理

1
2
3
4
5
6
7
8
9
10
11
# 添加虚拟主机
rabbitmqctl add_vhost /my_vhost

# 删除虚拟主机
rabbitmqctl delete_vhost /my_vhost

# 查看虚拟主机列表
rabbitmqctl list_vhosts

# 设置虚拟主机权限
rabbitmqctl set_permissions -p /my_vhost username ".*" ".*" ".*"

2. 队列和交换机管理

2.1. 队列操作

1
2
3
4
5
6
7
8
9
10
11
# 查看队列列表
rabbitmqctl list_queues

# 查看队列详细信息
rabbitmqctl list_queues name messages consumers memory

# 清空队列
rabbitmqctl purge_queue queue_name

# 删除队列
rabbitmqctl delete_queue queue_name

2.2. 交换机操作

1
2
3
4
5
6
7
8
# 查看交换机列表
rabbitmqctl list_exchanges

# 查看交换机详细信息
rabbitmqctl list_exchanges name type

# 删除交换机
rabbitmqctl delete_exchange exchange_name

2.3. 绑定管理

1
2
3
4
5
# 查看绑定关系
rabbitmqctl list_bindings

# 查看特定队列的绑定
rabbitmqctl list_bindings source_name destination_name

3. 消息操作

3.1. 消息查看

1
2
3
4
5
# 获取队列中的消息数量
rabbitmqctl list_queues name messages

# 查看消息详情（需要插件）
rabbitmqadmin get queue=/queue_name count=1

3.2. 消息移动

1
2
# 将消息从一个队列移动到另一个队列
rabbitmqctl move_messages source_queue destination_queue count=10

四、原理和代码示例

1. 核心组件

$2

routing_key=order.created

routing_key=order.paid

routing_key=log.error

routing_key=log.info

精确匹配 order.created

精确匹配 order.paid

匹配 log.*

匹配 log.error

Binding: order.created

Binding: order.paid

Binding: log.*

Binding: log.error

生产者1 订单服务

Direct Exchange

生产者2 日志服务

Topic Exchange

订单创建队列

订单支付队列

日志队列

错误日志队列

订单消费者1

订单消费者2

支付消费者1

日志消费者1

日志消费者2

错误日志消费者

1.1. 生产者(Producer)

  • 消息的发送方
  • 将消息发送到交换机

1.2. 消费者(Consumer)

  • 消息的接收方
  • 从队列中获取消息并处理

1.3. 交换机(Exchange)

  • 接收生产者发送的消息
  • 根据绑定规则将消息路由到队列

1.4. 队列(Queue)

  • 存储消息的容器
  • 消费者从队列中获取消息

1.5. 绑定(Binding)

  • 交换机和队列之间的关联关系
  • 包含路由键规则

1.6. 路由键(Routing Key)

  • 消息的路由规则
  • 交换机根据路由键决定消息发送到哪个队列

2. 交换机（Exchange）

2.1. 类型

1) Direct Exchange

  • 精确匹配路由键
  • 一对一关系
$2

发送消息 routing_key error

routing_key error

routing_key warning

routing_key info

生产者

Direct Exchange

Queue1: error队列

Queue2: warning队列

Queue3: info队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Go示例
err = ch.ExchangeDeclare(
    "direct_logs", // name
    "direct",      // type
    true,          // durable
    false,         // auto-deleted
    false,         // internal
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to declare an exchange")

err = ch.QueueBind(
    "queue1",      // queue name
    "error",       // routing key
    "direct_logs", // exchange
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to bind a queue")

err = ch.QueueBind(
    "queue2",      // queue name
    "warning",     // routing key
    "direct_logs", // exchange
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to bind a queue")

2) Fanout Exchange

  • 广播模式，忽略路由键
  • 一对多关系
$2

发送消息

广播到所有队列

广播到所有队列

广播到所有队列

生产者

Fanout Exchange

Queue1

Queue2

Queue3

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Go示例
err = ch.ExchangeDeclare(
    "fanout_logs", // name
    "fanout",      // type
    true,          // durable
    false,         // auto-deleted
    false,         // internal
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to declare an exchange")

err = ch.QueueBind(
    "queue1",      // queue name
    "",            // routing key (fanout模式忽略路由键)
    "fanout_logs", // exchange
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to bind a queue")

err = ch.QueueBind(
    "queue2",      // queue name
    "",            // routing key (fanout模式忽略路由键)
    "fanout_logs", // exchange
    false,         // no-wait
    nil,           // arguments
)
failOnError(err, "Failed to bind a queue")

3) Topic Exchange

  • 模式匹配路由键
  • 支持通配符: * (一个单词), # (零个或多个单词)
$2

发送消息 routing_key=kern.error

发送消息 routing_key=kern.warning

发送消息 routing_key=user.error

发送消息 routing_key=user.info

匹配 kern.*

匹配 *.error

匹配 user.#

生产者

Topic Exchange

Queue1: kern.*

Queue2: *.error

Queue3: user.#

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Go示例
err = ch.ExchangeDeclare(
    "topic_logs", // name
    "topic",      // type
    true,         // durable
    false,        // auto-deleted
    false,        // internal
    false,        // no-wait
    nil,          // arguments
)
failOnError(err, "Failed to declare an exchange")

err = ch.QueueBind(
    "queue1",     // queue name
    "*.error",    // routing key (*匹配一个单词)
    "topic_logs", // exchange
    false,        // no-wait
    nil,          // arguments
)
failOnError(err, "Failed to bind a queue")

err = ch.QueueBind(
    "queue2",     // queue name
    "kern.*",     // routing key (*匹配一个单词)
    "topic_logs", // exchange
    false,        // no-wait
    nil,          // arguments
)
failOnError(err, "Failed to bind a queue")

4) Headers Exchange

  • 基于消息头属性路由
  • 不使用路由键
$2

发送消息 headers: format=pdf, type=report

发送消息 headers: format=json, type=log

发送消息 headers: format=xml, type=data

匹配 format=pdf, type=report

匹配 format=json

匹配 type=data

生产者

Headers Exchange

Queue1: PDF报告

Queue2: JSON格式

Queue3: 数据类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// Go示例
err = ch.ExchangeDeclare(
    "headers_logs", // name
    "headers",      // type
    true,           // durable
    false,          // auto-deleted
    false,          // internal
    false,          // no-wait
    nil,            // arguments
)
failOnError(err, "Failed to declare an exchange")

err = ch.QueueBind(
    "queue1", // queue name
    "",       // routing key (headers模式忽略路由键)
    "headers_logs", // exchange
    false,    // no-wait
    amqp.Table{
        "format": "pdf",
        "type":   "report",
    }, // arguments
)
failOnError(err, "Failed to bind a queue")

2.2. 其他属性

1) durable

作用: 决定交换机是否持久化存储

  • true: 交换机持久化，RabbitMQ重启后仍然存在
  • false: 交换机临时存在，RabbitMQ重启后消失

应用场景:

  • 生产环境关键业务：在电商订单处理、支付系统等核心业务中，交换机必须持久化，确保RabbitMQ服务重启后交换机仍然存在，避免消息路由中断。
  • 系统可靠性要求高的场景：金融交易系统、银行核心系统等对数据一致性要求极高的场景，持久化交换机是基础保障。
  • 长时间运行的服务：7×24小时不间断运行的服务，如消息通知系统、日志收集系统等，需要保证交换机的持久性。
  • 避免重复配置：在复杂的微服务架构中，交换机配置复杂，持久化可以避免服务重启后需要重新创建和配置交换机。

2) auto-deleted

作用: 决定交换机是否在没有绑定队列时自动删除

  • true: 当最后一个绑定队列解绑后，交换机自动删除，注意：没有绑定过队列不会清理
  • false: 交换机需要手动删除

应用场景:

  • 临时任务处理：在批处理作业、定时任务等场景中，任务完成后交换机自动删除，避免资源浪费。
  • 动态服务发现：在微服务架构中，临时服务实例注册的交换机，服务下线后自动清理，保持系统整洁。
  • 测试环境：在单元测试、集成测试中，创建临时交换机进行测试，测试完成后自动删除，避免测试数据污染。
  • 短期会话管理：聊天应用、在线会议等场景中，会话相关的交换机在会话结束后自动删除。
  • 资源敏感环境：在资源受限的环境（如容器、Serverless）中，自动删除不使用的交换机，优化资源利用。

3) internal

作用: 决定交换机是否为内部交换机

  • true: 内部交换机，只能被其他交换机绑定，不能被生产者直接使用
  • false: 普通交换机，可以被生产者直接使用

内部交换机使用Bind函数将此交换机绑定到其他交换机的下一跳，一般用于一些内部敏感的交换机，避免外部直接访问。

4) no-wait

作用: 决定是否等待服务器确认操作完成

  • true: 异步操作，不等待服务器确认，立即返回
  • false: 同步操作，等待服务器确认后返回

当前操作是否等待，不等待则拿不到一些错误信息。一般不用，只有特定要求性能很高的场景可以自己保证一定成功的时候使用

3. 队列

4. 消息

五、小技巧与高级特性

1. 消息确认机制

1.1. 自动确认

1
2
3
4
5
6
7
8
9
10
// 消费者自动确认
msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    true,   // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
)

1.2. 手动确认

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func callback(d amqp.Delivery) {
    log.Printf(" [x] Received %s", d.Body)
    // 处理消息...
    d.Ack(false) // false表示只确认当前消息
}

msgs, err := ch.Consume(
    q.Name, // queue
    "",     // consumer
    false,  // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
)

1.3. 拒绝消息

1
2
3
4
5
6
7
8
9
10
11
func callback(d amqp.Delivery) {
    log.Printf(" [x] Received %s", d.Body)
    // 拒绝消息，不重新入队
    d.Reject(false)

    // 或者拒绝消息，重新入队
    d.Reject(true)

    // 批量拒绝（Nack）
    d.Nack(false, true) // multiple=false, requeue=true
}

2. 消息持久化

2.1. 队列持久化

1
2
3
4
5
6
7
8
9
// 声明持久化队列
q, err := ch.QueueDeclare(
    "durable_queue", // name
    true,            // durable
    false,           // delete when unused
    false,           // exclusive
    false,           // no-wait
    nil,             // arguments
)

2.2. 消息持久化

1
2
3
4
5
6
7
8
9
10
11
// 发送持久化消息
err = ch.Publish(
    "",     // exchange
    q.Name, // routing key
    false,  // mandatory
    false,  // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
        DeliveryMode: amqp.Persistent, // 持久化消息
    })

3. 消息优先级

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 声明支持优先级的队列
_, err = ch.QueueDeclare(
    "priority_queue", // name
    true,             // durable
    false,            // delete when unused
    false,            // exclusive
    false,            // no-wait
    amqp.Table{
        "x-max-priority": 10, // 设置最大优先级
    },
)
failOnError(err, "Failed to declare a queue")

// 发送带优先级的消息
err = ch.Publish(
    "",              // exchange
    "priority_queue", // routing key
    false,           // mandatory
    false,           // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(message),
        Priority:    5, // 设置优先级 (0-10)
    })
failOnError(err, "Failed to publish a message")

4. 死信队列

4.1. 配置死信队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// 声明死信交换机
err = ch.ExchangeDeclare(
    "dlx_exchange", // name
    "direct",       // type
    true,           // durable
    false,          // auto-deleted
    false,          // internal
    false,          // no-wait
    nil,            // arguments
)

// 声明死信队列
_, err = ch.QueueDeclare(
    "dlx_queue", // name
    true,        // durable
    false,       // delete when unused
    false,       // exclusive
    false,       // no-wait
    nil,         // arguments
)

// 绑定死信队列
err = ch.QueueBind(
    "dlx_queue",        // queue name
    "dlx_routing_key",  // routing key
    "dlx_exchange",     // exchange
    false,              // no-wait
    nil,                // arguments
)

// 声明主队列，设置死信参数
_, err = ch.QueueDeclare(
    "main_queue", // name
    true,         // durable
    false,        // delete when unused
    false,        // exclusive
    false,        // no-wait
    amqp.Table{
        "x-dead-letter-exchange":    "dlx_exchange",
        "x-dead-letter-routing-key": "dlx_routing_key",
        "x-message-ttl":             60000, // 消息TTL 60秒
    },
)

4.2. 死信队列使用场景

  • 消息过期
  • 队列长度限制
  • 消息被拒绝且不重新入队

5. 延迟队列

5.1. 使用TTL实现延迟队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 声明延迟队列
_, err = ch.QueueDeclare(
    "delay_queue", // name
    true,          // durable
    false,         // delete when unused
    false,         // exclusive
    false,         // no-wait
    amqp.Table{
        "x-message-ttl":             30000, // 30秒延迟
        "x-dead-letter-exchange":    "amq.direct",
        "x-dead-letter-routing-key": "target_queue",
    },
)
failOnError(err, "Failed to declare a queue")

// 声明目标队列
_, err = ch.QueueDeclare(
    "target_queue", // name
    true,           // durable
    false,          // delete when unused
    false,          // exclusive
    false,          // no-wait
    nil,            // arguments
)
failOnError(err, "Failed to declare a queue")

5.2. 使用延迟插件

1
2
# 安装延迟插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Go示例
// 声明延迟交换机
err = ch.ExchangeDeclare(
    "delayed_exchange", // name
    "x-delayed-message", // type
    true,               // durable
    false,              // auto-deleted
    false,              // internal
    false,              // no-wait
    amqp.Table{
        "x-delayed-type": "direct",
    },
)
failOnError(err, "Failed to declare an exchange")

// 发送延迟消息
err = ch.Publish(
    "delayed_exchange", // exchange
    "target_queue",     // routing key
    false,              // mandatory
    false,              // immediate
    amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(message),
        Headers: amqp.Table{
            "x-delay": 30000, // 30秒延迟
        },
    })
failOnError(err, "Failed to publish a message")

六、集群

1. 集群架构

1.1. 普通集群

[待补充mermaid图]

  • 队列在单个节点上，其他节点只有元数据
  • 节点故障时队列不可用

1.2. 镜像集群

[待补充mermaid图]

  • 队列在多个节点上复制
  • 高可用性，节点故障时自动切换

2. 搭建步骤

2.1. 准备工作

1
2
3
4
5
6
7
8
9
10
# 在所有节点上安装RabbitMQ
# 确保节点间可以通信
# 同步Erlang cookie

# 复制cookie文件
scp /var/lib/rabbitmq/.erlang.cookie node2:/var/lib/rabbitmq/
scp /var/lib/rabbitmq/.erlang.cookie node3:/var/lib/rabbitmq/

# 重启RabbitMQ服务
sudo systemctl restart rabbitmq-server

2.2. 组建集群

1
2
3
4
5
6
7
8
# 在node2和node3上执行，加入node1的集群
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

# 查看集群状态
rabbitmqctl cluster_status

2.3. 配置镜像队列

1
2
3
4
5
# 设置镜像策略
rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all","ha-sync-mode":"automatic"}'

# 查看策略
rabbitmqctl list_policies

3. 集群管理

3.1. 节点管理

1
2
3
4
5
6
7
8
9
# 查看集群状态
rabbitmqctl cluster_status

# 移除节点
rabbitmqctl forget_cluster_node rabbit@node2

# 改变节点类型
rabbitmqctl change_cluster_node_type disc
rabbitmqctl change_cluster_node_type ram

3.2. 集群同步

1
2
3
4
5
# 手动同步队列
rabbitmqctl sync_queue queue_name

# 取消同步
rabbitmqctl cancel_sync_queue queue_name

3.3. 集群脑裂

问题: 集群节点间通信中断，形成多个分区

原因:

  • 网络分区
  • 节点故障
  • 配置错误

解决方案:

1
2
3
4
5
6
7
8
9
# 配置自动恢复
cluster_partition_handling = autoheal

# 手动恢复集群
rabbitmqctl force_boot
rabbitmqctl start_app

# 监控集群状态
rabbitmqctl cluster_status

七、监控和运维

1. 管理界面

1.1. 访问管理界面

1.2. 主要功能

  • 连接管理
  • 通道管理
  • 交换机管理
  • 队列管理
  • 消息查看
  • 用户管理
  • 策略管理

2. 命令行监控

2.1. 基础监控

1
2
3
4
5
6
7
8
9
10
11
# 查看连接数
rabbitmqctl list_connections

# 查看通道数
rabbitmqctl list_channels

# 查看消费者
rabbitmqctl list_consumers

# 查看内存使用
rabbitmqctl status

2.2. 性能监控

1
2
3
4
5
6
7
8
# 查看队列消息速率
rabbitmqctl list_queues name messages_ready messages_unacknowledged

# 查看交换机消息速率
rabbitmqctl list_exchanges name message_stats

# 查看节点性能
rabbitmqctl node_health_check

3. 日志管理

3.1. 日志配置

1
2
3
4
5
6
7
# 日志级别
log.file.level = info
log.console = true

# 日志文件
log.file = /var/log/rabbitmq/rabbit.log
log.exchange.file = /var/log/rabbitmq/rabbit_exchange.log

3.2. 日志分析

1
2
3
4
5
6
7
8
# 查看错误日志
grep -i error /var/log/rabbitmq/rabbit.log

# 查看连接日志
grep -i connection /var/log/rabbitmq/rabbit.log

# 实时查看日志
tail -f /var/log/rabbitmq/rabbit.log

4. 性能调优

4.1. 内存优化

1
2
3
4
5
# 调整内存高水位
vm_memory_high_watermark.relative = 0.7

# 启用内存警报
memory_monitor_interval = 1000

4.2. 磁盘优化

1
2
3
4
5
# 调整磁盘空间限制
disk_free_limit.relative = 1.5

# 启用磁盘监控
disk_monitor_interval = 10000

4.3. 网络优化

1
2
3
4
5
6
# 调整TCP缓冲区
tcp_listen_options.backlog = 128
tcp_listen_options.nodelay = true

# 调整心跳间隔
heartbeat = 60

八、实战

1. 最简单的rabbitmq的客户端和服务端实现

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
package main

import (
	"log"

	"github.com/streadway/amqp"
)

// 错误处理函数
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	// 连接到RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "无法连接到RabbitMQ服务器")
	defer conn.Close()

	// 创建一个通道
	ch, err := conn.Channel()
	failOnError(err, "无法创建通道")
	defer ch.Close()

	// 声明一个队列
	q, err := ch.QueueDeclare(
		"hello", // 队列名称
		false,   // 持久化
		false,   // 非自动删除
		false,   // 非排他性
		false,   // 不阻塞
		nil,     // 额外参数
	)
	failOnError(err, "无法声明队列")

	// 要发送的消息
	body := "Hello World!"
	// 发送消息到队列
	err = ch.Publish(
		"",     // 交换机
		q.Name, // 路由键（队列名称）
		false,  // 强制的
		false,  // 立即的
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
		})
	failOnError(err, "无法发送消息")

	log.Printf(" [x] 已发送 %s", body)
}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
package main

import (
	"log"

	"github.com/streadway/amqp"
)

// 错误处理函数
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	// 连接到RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "无法连接到RabbitMQ服务器")
	defer conn.Close()

	// 创建一个通道
	ch, err := conn.Channel()
	failOnError(err, "无法创建通道")
	defer ch.Close()

	// 声明队列（需要与生产者使用相同的队列名称）
	q, err := ch.QueueDeclare(
		"hello", // 队列名称
		false,   // 持久化
		false,   // 非自动删除
		false,   // 非排他性
		false,   // 不阻塞
		nil,     // 额外参数
	)
	failOnError(err, "无法声明队列")

	// 消费消息
	msgs, err := ch.Consume(
		q.Name, // 队列名称
		"",     // 消费者标签
		true,   // 自动确认
		false,  // 非排他性
		false,  // 不局部
		false,  // 不阻塞
		nil,    // 额外参数
	)
	failOnError(err, "无法注册消费者")

	// 创建一个通道来接收退出信号
	forever := make(chan bool)

	// 启动一个goroutine处理接收到的消息
	go func() {
		for d := range msgs {
			log.Printf(" [x] 接收到 %s", d.Body)
		}
	}()

	log.Printf(" [*] 等待消息中。按CTRL+C退出")
	<-forever // 阻塞，保持程序运行
}

1. 最佳实践

1.1. 消息设计

  • 使用有意义的队列名称
  • 合理设置消息TTL
  • 使用适当的交换机类型
  • 避免过大的消息体

1.2. 错误处理

  • 实现消息确认机制
  • 使用死信队列处理失败消息
  • 实现重试机制
  • 记录详细的错误日志

1.3. 性能优化

  • 使用连接池
  • 批量发送消息
  • 合理设置预取计数
  • 监控系统资源使用

1.4. 工程化

1.5. 安全

[待补充]

2. 防止消息重复消费

[待补充]

3. 排查消息堆积

[待补充]

4. 如何保证消息顺序性（正常、重试、死信队列情况下）

[待补充]

九、踩坑记

1. 消息问题

1.1. 消息堆积

1
2
3
4
5
6
7
8
# 查看队列消息数量
rabbitmqctl list_queues name messages

# 增加消费者
# 检查消费者处理速度

# 清空队列（谨慎操作）
rabbitmqctl purge_queue queue_name

1.2. 消息丢失

1
2
3
4
5
6
7
8
# 检查队列持久化设置
rabbitmqctl list_queues name durable

# 检查消息持久化设置
# 确保生产者设置了delivery_mode=2

# 检查消费者确认机制
# 确保消费者正确处理消息确认

2. 性能问题

2.1. 内存使用过高

1
2
3
4
5
6
7
8
# 查看内存使用
rabbitmqctl status

# 调整内存配置
vm_memory_high_watermark.relative = 0.6

# 清理未使用的队列
rabbitmqctl list_queues name messages consumers

2.2. 磁盘空间不足

1
2
3
4
5
6
7
8
# 查看磁盘使用
df -h

# 调整磁盘配置
disk_free_limit.relative = 1.0

# 清理日志文件
# 调整日志级别