RabbitMQ

RabbitMQ
	文件:RabbitMQ logo.svg
Developer(s)	Pivotal Software (现为VMware Tanzu)
Initial release	2007年
	模板:Infobox software/simple
Repository	{{URL\|example.com\|optional display text}};
Written in	Erlang
Engine
Operating system	跨平台
Type	消息代理
License	Mozilla Public License
Website	https://www.rabbitmq.com

RabbitMQ是一个开源的消息队列系统，实现了高级消息队列协议(AMQP)标准。作为轻量级的中间件，它支持多种消息协议，能在分布式系统中可靠地传递消息，广泛应用于微服务架构、异步任务处理和系统解耦等场景。

核心概念[编辑 | 编辑源代码]

基本组件[编辑 | 编辑源代码]

RabbitMQ的核心架构包含以下关键组件：

生产者(Producer)：发送消息的应用程序
消费者(Consumer)：接收消息的应用程序
队列(Queue)：存储消息的缓冲区
交换机(Exchange)：接收生产者消息并根据规则路由到队列
绑定(Binding)：连接交换机和队列的规则
虚拟主机(Virtual Host)：独立的隔离环境

消息流[编辑 | 编辑源代码]

交换机类型[编辑 | 编辑源代码]

RabbitMQ提供四种主要交换机类型：

直连交换机(Direct)[编辑 | 编辑源代码]

根据路由键(routing key)精确匹配队列
适用于单播消息路由

扇形交换机(Fanout)[编辑 | 编辑源代码]

广播消息到所有绑定队列
忽略路由键
适用于发布/订阅场景

主题交换机(Topic)[编辑 | 编辑源代码]

基于模式匹配路由键
支持通配符(*匹配一个单词，#匹配零或多个单词)
适用于多条件路由

头交换机(Headers)[编辑 | 编辑源代码]

基于消息头属性而非路由键
可设置匹配条件(x-match: all/any)

安装与配置[编辑 | 编辑源代码]

安装方法[编辑 | 编辑源代码]

在Ubuntu系统上安装RabbitMQ：

# 添加RabbitMQ仓库
sudo apt-get install -y curl gnupg
curl -fsSL https://github.com/rabbitmq/signing-keys/releases/download/2.0/rabbitmq-release-signing-key.asc | sudo gpg --dearmor -o /usr/share/keyrings/rabbitmq-archive-keyring.gpg

# 安装服务
echo "deb [signed-by=/usr/share/keyrings/rabbitmq-archive-keyring.gpg] https://dl.bintray.com/rabbitmq-erlang/debian $(lsb_release -sc) main" | sudo tee /etc/apt/sources.list.d/rabbitmq.list
sudo apt-get update
sudo apt-get install -y rabbitmq-server

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

基本管理[编辑 | 编辑源代码]

常用管理命令：

# 查看状态
sudo rabbitmqctl status

# 添加用户
sudo rabbitmqctl add_user myuser mypassword

# 设置权限
sudo rabbitmqctl set_permissions -p / myuser ".*" ".*" ".*"

# 启用管理插件
sudo rabbitmq-plugins enable rabbitmq_management

客户端开发[编辑 | 编辑源代码]

Python示例[编辑 | 编辑源代码]

使用pika库发送消息：

import pika

# 建立连接
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='hello')

# 发送消息
channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='Hello World!')
print(" [x] Sent 'Hello World!'")

connection.close()

接收消息的消费者代码：

import pika

def callback(ch, method, properties, body):
    print(f" [x] Received {body}")

connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')
channel.basic_consume(
    queue='hello',
    auto_ack=True,
    on_message_callback=callback)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()