07-大型项目架构/02-消息队列/01-RabbitMQ深入
RabbitMQ 深入
学习目标
- 掌握 RabbitMQ 核心概念和架构
- 理解生产者消费者模式
- 学会交换机类型（Direct、Topic、Fanout）
- 实现可靠消息生产和消费
前置知识
- AMQP 协议基础
- 异步编程概念
- 微服务架构基础
知识点列表
1. RabbitMQ 架构概述
1.1 核心概念
// rabbitmq/concepts.js
const RabbitMQConcepts = {
// 生产者 (Producer)
producer: {
description: '消息发送者',
role: '发布消息到交换机',
example: '订单服务发布订单创建消息'
},
// 消费者 (Consumer)
consumer: {
description: '消息接收者',
role: '从队列消费消息',
example: '库存服务消费订单消息'
},
// 队列 (Queue)
queue: {
description: '消息存储缓冲区',
properties: ['名称', '持久化', '独占', '自动删除'],
example: 'order.queue'
},
// 交换机 (Exchange)
exchange: {
description: '消息路由',
types: ['Direct', 'Topic', 'Fanout', 'Headers'],
example: 'order.exchange'
},
// 绑定 (Binding)
binding: {
description: '队列与交换机的关联',
key: '路由键',
example: 'order.created -> order.queue'
}
};
1.2 安装与连接
# Docker 安装
docker run -d –name rabbitmq \\
-p 5672:5672 \\
-p 15672:15672 \\
-e RABBITMQ_DEFAULT_USER=admin \\
-e RABBITMQ_DEFAULT_PASS=admin \\
rabbitmq:3.12-management
// connection.js
const amqp = require('amqplib');
class RabbitMQConnection {
constructor(config) {
this.url = config.url || 'amqp://localhost';
this.connection = null;
this.channel = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = config.maxReconnectAttempts || 10;
this.reconnectDelay = config.reconnectDelay || 5000;
}
async connect() {
try {
this.connection = await amqp.connect(this.url);
this.channel = await this.connection.createChannel();
console.log('RabbitMQ 连接成功');
// 连接事件监听
this.connection.on('error', (err) => {
console.error('RabbitMQ 连接错误:', err);
this.reconnect();
});
this.connection.on('close', () => {
console.log('RabbitMQ 连接关闭');
this.reconnect();
});
this.reconnectAttempts = 0;
return this.channel;
} catch (err) {
console.error('RabbitMQ 连接失败:', err.message);
this.reconnect();
throw err;
}
}
async reconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error('达到最大重连次数，停止重连');
return;
}
this.reconnectAttempts++;
console.log(`尝试重连 (${this.reconnectAttempts}/${this.maxReconnectAttempts})…`);
setTimeout(() => this.connect(), this.reconnectDelay);
}
async close() {
await this.channel?.close();
await this.connection?.close();
}
getChannel() {
return this.channel;
}
}
module.exports = RabbitMQConnection;
2. 生产者消费者模式
2.1 简单队列模式
// simple-queue/producer.js
class SimpleProducer {
constructor(channel, queueName) {
this.channel = channel;
this.queueName = queueName;
}
async setup() {
// 声明队列（持久化）
await this.channel.assertQueue(this.queueName, {
durable: true,
exclusive: false,
autoDelete: false
});
}
async send(message, options = {}) {
const content = Buffer.from(JSON.stringify(message));
this.channel.sendToQueue(this.queueName, content, {
persistent: options.persistent !== false,
expiration: options.expiration,
headers: {
'x-message-type': message.type,
'x-timestamp': Date.now(),
…options.headers
}
});
console.log(`消息已发送: ${this.queueName}`, message);
return { success: true, messageId: message.id };
}
async sendBatch(messages) {
const promises = messages.map(msg => this.send(msg));
await Promise.all(promises);
console.log(`批量发送 ${messages.length} 条消息`);
}
}
// simple-queue/consumer.js
class SimpleConsumer {
constructor(channel, queueName, handler) {
this.channel = channel;
this.queueName = queueName;
this.handler = handler;
this.isRunning = false;
}
async setup() {
await this.channel.assertQueue(this.queueName, { durable: true });
// 每次只取一条消息
await this.channel.prefetch(1);
}
async start() {
this.isRunning = true;
console.log(`开始消费队列: ${this.queueName}`);
this.channel.consume(this.queueName, async (msg) => {
if (!msg) return;
try {
const message = JSON.parse(msg.content.toString());
console.log(`收到消息: ${this.queueName}`, message);
await this.handler(message, msg);
// 确认消息已处理
this.channel.ack(msg);
console.log(`消息处理成功: ${message.id}`);
} catch (err) {
console.error(`消息处理失败:`, err);
// 拒绝消息并决定是否重新入队
const shouldRequeue = this.shouldRequeue(msg, err);
this.channel.nack(msg, false, shouldRequeue);
}
});
}
stop() {
this.isRunning = false;
this.channel.cancel(this.queueName);
}
shouldRequeue(msg, error) {
const retryCount = msg.properties.headers?.['x-retry-count'] || 0;
const maxRetries = 3;
// 超过重试次数则不再重新入队
return retryCount < maxRetries;
}
}
// 使用示例
async function demoSimpleQueue() {
const connection = new RabbitMQConnection({ url: 'amqp://localhost' });
const channel = await connection.connect();
// 生产者
const producer = new SimpleProducer(channel, 'task.queue');
await producer.setup();
// 消费者
const consumer = new SimpleConsumer(channel, 'task.queue', async (message) => {
console.log('处理任务:', message);
// 模拟处理
await new Promise(resolve => setTimeout(resolve, 1000));
});
await consumer.setup();
await consumer.start();
// 发送消息
await producer.send({
id: Date.now(),
type: 'task',
data: { name: 'task1' }
});
}
2.2 工作队列模式
// work-queue/producer.js
class WorkQueueProducer {
constructor(channel, queueName) {
this.channel = channel;
this.queueName = queueName;
}
async setup() {
await this.channel.assertQueue(this.queueName, { durable: true });
}
async addTask(task, priority = 0) {
const message = {
id: this.generateTaskId(),
task,
priority,
createdAt: Date.now()
};
this.channel.sendToQueue(this.queueName, Buffer.from(JSON.stringify(message)), {
persistent: true,
priority,
headers: { 'x-priority': priority }
});
console.log(`任务已添加: ${message.id}, 优先级: ${priority}`);
return message.id;
}
generateTaskId() {
return `task-${Date.now()}–${Math.random().toString(36)}`;
}
}
// work-queue/worker.js
class WorkQueueWorker {
constructor(channel, queueName, handler, workerId) {
this.channel = channel;
this.queueName = queueName;
this.handler = handler;
this.workerId = workerId;
this.stats = {
processed: 0,
failed: 0,
lastProcessed: null
};
}
async setup() {
// 声明优先级队列
await this.channel.assertQueue(this.queueName, {
durable: true,
maxPriority: 10
});
// 每次只取一条消息
await this.channel.prefetch(1);
}
async start() {
console.log(`Worker ${this.workerId} 启动`);
this.channel.consume(this.queueName, async (msg) => {
if (!msg) return;
const startTime = Date.now();
const task = JSON.parse(msg.content.toString());
try {
console.log(`Worker ${this.workerId} 处理任务: ${task.id}`);
await this.handler(task);
this.channel.ack(msg);
this.stats.processed++;
this.stats.lastProcessed = Date.now();
const duration = Date.now() – startTime;
console.log(`Worker ${this.workerId} 完成任务: ${task.id}, 耗时: ${duration}ms`);
} catch (err) {
console.error(`Worker ${this.workerId} 处理失败:`, err);
const retryCount = (msg.properties.headers?.['x-retry-count'] || 0) + 1;
const maxRetries = 3;
if (retryCount <= maxRetries) {
// 重新入队，延迟重试
setTimeout(() => {
this.channel.nack(msg, false, true);
}, retryCount * 1000);
} else {
this.channel.nack(msg, false, false);
this.stats.failed++;
}
}
});
}
getStats() {
return { …this.stats, workerId: this.workerId };
}
}
// 工作队列管理器
class WorkQueueManager {
constructor(channel, queueName, workerCount = 3) {
this.channel = channel;
this.queueName = queueName;
this.workerCount = workerCount;
this.workers = [];
this.producer = new WorkQueueProducer(channel, queueName);
}
async start() {
await this.producer.setup();
for (let i = 0; i < this.workerCount; i++) {
const worker = new WorkQueueWorker(
this.channel,
this.queueName,
this.processTask.bind(this),
`worker-${i + 1}`
);
await worker.setup();
await worker.start();
this.workers.push(worker);
}
console.log(`工作队列已启动，${this.workerCount} 个 worker`);
}
async processTask(task) {
// 模拟不同处理时间
const duration = task.task.duration || 1000;
await new Promise(resolve => setTimeout(resolve, duration));
return { success: true, processedAt: Date.now() };
}
async addTask(task, priority = 0) {
return this.producer.addTask(task, priority);
}
getStats() {
return this.workers.map(w => w.getStats());
}
}
3. 交换机类型
3.1 Direct Exchange（直连交换机）
// exchanges/direct-exchange.js
class DirectExchange {
constructor(channel, exchangeName) {
this.channel = channel;
this.exchangeName = exchangeName;
this.queues = new Map();
}
async setup() {
await this.channel.assertExchange(this.exchangeName, 'direct', {
durable: true,
autoDelete: false
});
console.log(`Direct Exchange 创建: ${this.exchangeName}`);
}
// 创建队列并绑定
async bindQueue(queueName, routingKey) {
await this.channel.assertQueue(queueName, { durable: true });
await this.channel.bindQueue(queueName, this.exchangeName, routingKey);
this.queues.set(routingKey, queueName);
console.log(`队列绑定: ${queueName} <- ${routingKey}`);
}
// 发送消息
async publish(routingKey, message, options = {}) {
const content = Buffer.from(JSON.stringify(message));
this.channel.publish(this.exchangeName, routingKey, content, {
persistent: options.persistent !== false,
headers: {
'x-routing-key': routingKey,
'x-timestamp': Date.now(),
…options.headers
}
});
console.log(`消息发布: ${this.exchangeName}:${routingKey}`);
return true;
}
// 消费消息
async consume(routingKey, handler) {
const queueName = this.queues.get(routingKey);
if (!queueName) {
throw new Error(`队列未绑定: ${routingKey}`);
}
await this.channel.prefetch(1);
this.channel.consume(queueName, async (msg) => {
if (msg) {
const message = JSON.parse(msg.content.toString());
await handler(message, routingKey);
this.channel.ack(msg);
}
});
console.log(`开始消费: ${routingKey}`);
}
}
// 使用示例：日志系统
async function demoDirectExchange() {
const connection = new RabbitMQConnection({ url: 'amqp://localhost' });
const channel = await connection.connect();
const exchange = new DirectExchange(channel, 'logs.direct');
await exchange.setup();
// 创建不同级别的队列
await exchange.bindQueue('error.queue', 'error');
await exchange.bindQueue('warning.queue', 'warning');
await exchange.bindQueue('info.queue', 'info');
// 错误处理器（只接收 error 级别）
await exchange.consume('error', async (message) => {
console.error('[ERROR]', message);
// 发送告警
});
// 警告处理器
await exchange.consume('warning', async (message) => {
console.warn('[WARNING]', message);
});
// 信息处理器
await exchange.consume('info', async (message) => {
console.log('[INFO]', message);
});
// 发送不同级别的日志
await exchange.publish('error', {
level: 'error',
message: '数据库连接失败',
timestamp: Date.now()
});
await exchange.publish('warning', {
level: 'warning',
message: '内存使用率过高',
usage: '85%'
});
await exchange.publish('info', {
level: 'info',
message: '用户登录成功',
userId: 123
});
}
3.2 Topic Exchange（主题交换机）
// exchanges/topic-exchange.js
class TopicExchange {
constructor(channel, exchangeName) {
this.channel = channel;
this.exchangeName = exchangeName;
this.bindings = new Map();
}
async setup() {
await this.channel.assertExchange(this.exchangeName, 'topic', {
durable: true
});
console.log(`Topic Exchange 创建: ${this.exchangeName}`);
}
// 绑定队列（支持通配符）
// * 匹配一个单词
// # 匹配零个或多个单词
async bindQueue(queueName, pattern) {
await this.channel.assertQueue(queueName, { durable: true });
await this.channel.bindQueue(queueName, this.exchangeName, pattern);
this.bindings.set(pattern, queueName);
console.log(`队列绑定: ${queueName} <- ${pattern}`);
}
async publish(routingKey, message) {
const content = Buffer.from(JSON.stringify(message));
this.channel.publish(this.exchangeName, routingKey, content, {
persistent: true,
headers: { 'x-routing-key': routingKey }
});
console.log(`消息发布: ${this.exchangeName}:${routingKey}`);
}
async subscribe(pattern, handler) {
const queueName = this.bindings.get(pattern);
if (!queueName) {
throw new Error(`未找到匹配的队列: ${pattern}`);
}
this.channel.consume(queueName, async (msg) => {
if (msg) {
const message = JSON.parse(msg.content.toString());
const routingKey = msg.fields.routingKey;
await handler(message, routingKey);
this.channel.ack(msg);
}
});
console.log(`订阅: ${pattern}`);
}
}
// 使用示例：事件驱动架构
async function demoTopicExchange() {
const connection = new RabbitMQConnection({ url: 'amqp://localhost' });
const channel = await connection.connect();
const exchange = new TopicExchange(channel, 'events.topic');
await exchange.setup();
// 用户事件订阅
await exchange.bindQueue('user.events.queue', 'user.*');
await exchange.bindQueue('user.created.queue', 'user.created');
await exchange.bindQueue('user.updated.queue', 'user.updated');
// 订单事件订阅
await exchange.bindQueue('order.events.queue', 'order.*');
await exchange.bindQueue('order.payment.queue', 'order.payment.*');
// 所有事件
await exchange.bindQueue('all.events.queue', '#');
// 订阅用户事件
await exchange.subscribe('user.*', async (message, routingKey) => {
console.log(`用户事件 [${routingKey}]:`, message);
});
// 订阅订单支付事件
await exchange.subscribe('order.payment.*', async (message, routingKey) => {
console.log(`支付事件 [${routingKey}]:`, message);
// 处理支付成功/失败
});
// 发布各种事件
await exchange.publish('user.created', {
userId: 123,
username: 'Alice',
timestamp: Date.now()
});
await exchange.publish('user.updated', {
userId: 123,
changes: { email: 'alice@example.com' }
});
await exchange.publish('order.created', {
orderId: 'ORD-001',
userId: 123,
total: 99.99
});
await exchange.publish('order.payment.success', {
orderId: 'ORD-001',
paymentId: 'PAY-001',
amount: 99.99
});
}
3.3 Fanout Exchange（扇出交换机）
// exchanges/fanout-exchange.js
class FanoutExchange {
constructor(channel, exchangeName) {
this.channel = channel;
this.exchangeName = exchangeName;
this.queues = new Set();
}
async setup() {
await this.channel.assertExchange(this.exchangeName, 'fanout', {
durable: true
});
console.log(`Fanout Exchange 创建: ${this.exchangeName}`);
}
// 创建临时队列并绑定
async bindTemporaryQueue() {
const queue = await this.channel.assertQueue('', { exclusive: true });
await this.channel.bindQueue(queue.queue, this.exchangeName, '');
this.queues.add(queue.queue);
console.log(`临时队列绑定: ${queue.queue}`);
return queue.queue;
}
// 创建持久化队列并绑定
async bindPersistentQueue(queueName) {
await this.channel.assertQueue(queueName, { durable: true });
await this.channel.bindQueue(queueName, this.exchangeName, '');
this.queues.add(queueName);
console.log(`持久队列绑定: ${queueName}`);
}
async broadcast(message) {
const content = Buffer.from(JSON.stringify(message));
this.channel.publish(this.exchangeName, '', content, {
persistent: true
});
console.log(`消息广播: ${this.exchangeName}`);
}
async consume(queueName, handler) {
this.channel.consume(queueName, async (msg) => {
if (msg) {
const message = JSON.parse(msg.content.toString());
await handler(message);
this.channel.ack(msg);
}
});
console.log(`开始消费: ${queueName}`);
}
}
// 使用示例：广播通知系统
async function demoFanoutExchange() {
const connection = new RabbitMQConnection({ url: 'amqp://localhost' });
const channel = await connection.connect();
const exchange = new FanoutExchange(channel, 'notifications.fanout');
await exchange.setup();
// 短信通知服务
await exchange.bindPersistentQueue('sms.notifications');
await exchange.consume('sms.notifications', async (message) => {
console.log(`[SMS] 发送短信: ${message.content}`, message.recipients);
// 调用短信服务
});
// 邮件通知服务
await exchange.bindPersistentQueue('email.notifications');
await exchange.consume('email.notifications', async (message) => {
console.log(`[EMAIL] 发送邮件: ${message.content}`, message.recipients);
// 调用邮件服务
});
// Webhook 服务（临时队列）
const webhookQueue = await exchange.bindTemporaryQueue();
await exchange.consume(webhookQueue, async (message) => {
console.log(`[WEBHOOK] 触发 Webhook: ${message.webhookUrl}`, message.data);
// 调用 Webhook
});
// 日志记录服务
await exchange.bindPersistentQueue('audit.logs');
await exchange.consume('audit.logs', async (message) => {
console.log(`[AUDIT] 记录审计日志:`, message);
// 写入审计日志
});
// 广播系统通知
await exchange.broadcast({
type: 'system.notification',
title: '系统维护通知',
content: '系统将于今晚 22:00 进行维护',
recipients: ['all'],
webhookUrl: 'https://hooks.slack.com/xxx',
timestamp: Date.now()
});
// 广播用户通知
await exchange.broadcast({
type: 'user.notification',
title: '订单状态更新',
content: '您的订单已发货',
userId: 123,
timestamp: Date.now()
});
}
4. 可靠消息模式
4.1 消息确认机制
// reliable/producer.js
class ReliableProducer {
constructor(channel, exchangeName) {
this.channel = channel;
this.exchangeName = exchangeName;
this.confirmChannel = null;
}
async setup() {
// 使用确认模式
this.confirmChannel = await this.channel.confirm();
await this.confirmChannel.assertExchange(this.exchangeName, 'topic', {
durable: true
});
}
async publishWithConfirm(routingKey, message) {
return new Promise((resolve, reject) => {
const content = Buffer.from(JSON.stringify(message));
this.confirmChannel.publish(
this.exchangeName,
routingKey,
content,
{ persistent: true },
(err, ok) => {
if (err) {
reject(err);
} else {
resolve(ok);
}
}
);
this.confirmChannel.waitForConfirms()
.then(() => {
console.log(`消息已确认: ${routingKey}`);
})
.catch(reject);
});
}
async publishBatch(messages) {
for (const msg of messages) {
const content = Buffer.from(JSON.stringify(msg.message));
this.confirmChannel.publish(
this.exchangeName,
msg.routingKey,
content,
{ persistent: true }
);
}
await this.confirmChannel.waitForConfirms();
console.log(`批量消息已确认: ${messages.length} 条`);
}
}
// reliable/consumer.js
class ReliableConsumer {
constructor(channel, queueName, handler) {
this.channel = channel;
this.queueName = queueName;
this.handler = handler;
}
async setup() {
await this.channel.assertQueue(this.queueName, { durable: true });
await this.channel.prefetch(1);
}
async start() {
this.channel.consume(this.queueName, async (msg) => {
if (!msg) return;
try {
const message = JSON.parse(msg.content.toString());
await this.handler(message);
// 手动确认
this.channel.ack(msg);
console.log(`消息已确认: ${message.id}`);
} catch (err) {
console.error(`消息处理失败:`, err);
// 拒绝并重新入队
this.channel.nack(msg, false, true);
}
});
}
}
4.2 死信队列
// dlx/dead-letter-queue.js
class DeadLetterExchange {
constructor(channel, mainExchange, dlxExchange) {
this.channel = channel;
this.mainExchange = mainExchange;
this.dlxExchange = dlxExchange;
}
async setup() {
// 创建死信交换机
await this.channel.assertExchange(this.dlxExchange, 'direct', {
durable: true
});
// 创建死信队列
const dlxQueue = await this.channel.assertQueue(`${this.dlxExchange}.queue`, {
durable: true
});
await this.channel.bindQueue(dlxQueue.queue, this.dlxExchange, 'dead');
// 创建主队列（带死信配置）
const mainQueue = await this.channel.assertQueue(`${this.mainExchange}.queue`, {
durable: true,
arguments: {
'x-dead-letter-exchange': this.dlxExchange,
'x-dead-letter-routing-key': 'dead',
'x-message-ttl': 60000, // 消息存活时间
'x-max-length': 10000, // 队列最大长度
'x-max-priority': 10 // 优先级
}
});
await this.channel.bindQueue(mainQueue.queue, this.mainExchange, '#');
console.log(`死信队列配置完成: ${this.mainExchange}`);
return { mainQueue, dlxQueue };
}
async processDeadLetters(handler) {
const queueName = `${this.dlxExchange}.queue`;
this.channel.consume(queueName, async (msg) => {
if (msg) {
const message = JSON.parse(msg.content.toString());
const reason = msg.properties.headers?.['x-death']?.[0]?.reason;
console.log(`死信消息: ${message.id}, 原因: ${reason}`);
await handler(message, reason);
this.channel.ack(msg);
}
});
console.log(`死信队列处理器已启动: ${queueName}`);
}
}
// 使用示例
async function demoDeadLetterQueue() {
const connection = new RabbitMQConnection({ url: 'amqp://localhost' });
const channel = await connection.connect();
const dlx = new DeadLetterExchange(
channel,
'order.process',
'order.dlx'
);
await dlx.setup();
// 处理死信消息
await dlx.processDeadLetters(async (message, reason) => {
console.log(`处理死信消息: ${message.id}, 原因: ${reason}`);
// 记录到数据库
await saveFailedMessage(message, reason);
// 发送告警
await sendAlert({
type: 'dead_letter',
message: message,
reason
});
});
}
练习题
基础题
进阶题
挑战题
练习题参考答案
基础题1
// 生产者
await channel.sendToQueue('queue', Buffer.from('message'));
// 消费者
channel.consume('queue', (msg) => {
console.log(msg.content.toString());
channel.ack(msg);
});
基础题2
const exchange = new DirectExchange(channel, 'logs');
await exchange.bindQueue('error.queue', 'error');
await exchange.publish('error', { level: 'error', msg: 'Error occurred' });
基础题3
const exchange = new TopicExchange(channel, 'events');
await exchange.bindQueue('user.queue', 'user.*');
await exchange.publish('user.created', { userId: 123 });
