RabbitMQ

RabbitMQ 简介

RabbitMQ 是一个开源的 消息代理（Message Broker），基于 AMQP（Advanced Message Queuing Protocol） 协议实现，由 Pivotal 维护。它被广泛用于 异步任务处理、应用解耦、流量削峰 等场景，支持多种消息模式（如点对点、发布/订阅），并具备高可用、可扩展的特性。RabbitMQ 以其 轻量级、易用性和丰富的插件生态 成为企业级消息队列的首选之一。

核心特性

1. 多种消息模式支持
   • 点对点（Queue）：单生产者 → 单消费者
   • 发布/订阅（Exchange）：支持 fanout、direct、topic、headers 四种路由规则
   • RPC（远程调用）：支持请求-响应模式
2. 高可用与持久化
   • 支持 镜像队列（Mirrored Queues） 实现数据冗余
   • 消息可持久化到磁盘（防止服务崩溃丢失）
3. 灵活的路由机制
   • Exchange（交换机） 负责消息路由
   • 支持 绑定键（Binding Key） 和 路由键（Routing Key） 动态匹配
4. 丰富的插件生态
   • Management Plugin：提供 Web 管理界面
   • MQTT Plugin：支持物联网协议
   • STOMP Plugin：兼容 WebSocket 通信
5. 多语言支持
   • 官方提供 Java、Python、.NET、Go、Node.js 等客户端 SDK
6. 轻量级与低延迟
   • 单节点可支持 万级 TPS，适合中小规模应用

核心概念

典型应用场景

1. 异步任务处理
   • 耗时操作（如邮件发送、图片处理）异步化，提升响应速度
2. 应用解耦
   • 微服务间通过消息通信，降低直接依赖（如订单服务 → 库存服务）
3. 流量削峰
   • 突发流量缓冲（如秒杀活动），避免系统过载
4. 日志收集
   • 多个应用日志统一发送到 RabbitMQ，由消费者集中处理
5. 事件驱动架构（EDA）
   • 系统状态变更通过消息广播（如用户注册成功通知）

快速入门示例

1. 安装与启动

# 使用 Docker 快速启动
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

访问 http://localhost:15672（默认账号：guest/guest）

2. Python 生产者示例

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 RabbitMQ!')
print(" [x] Sent 'Hello RabbitMQ!'")
connection.close()

3. Java 消费者示例

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

channel.queueDeclare("hello", false, false, false, null);
channel.basicConsume("hello", true, (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println(" [x] Received '" + message + "'");
}, consumerTag -> {});

性能优化建议

1. 集群部署
   • 使用 镜像队列 实现高可用（ha-mode=all）
   • 多节点部署提升吞吐量
2. 消息设计
   • 避免大消息（建议 **<1MB**），可拆分或存储引用（如文件ID）
   • 启用 消息压缩（如 gzip）减少网络传输
3. 资源调优
   • 调整 channel_max 和 frame_max 提升并发能力
   • 监控内存和磁盘使用（rabbitmqctl status）
4. 消费者优化
   • 使用 basicQos 限制预取数量（防止单消费者过载）
   • 多线程消费者提升处理速度