Node.js 消息队列集成实战：RabbitMQ、Kafka、BullMQ 与事件驱动架构

// Node.js Stream 中的背压处理
const readable = getMessageStream();
const writable = getConsumerStream();

readable.on('data', (chunk) => {
    const canContinue = writable.write(chunk);
    if (!canContinue) {
        readable.pause();                    // 暂停生产
        writable.once('drain', () => {
            readable.resume();               // 恢复生产
        });
    }
});

const amqp = require('amqplib');

// 连接到 RabbitMQ
const conn = await amqp.connect('amqp://guest:guest@localhost:5672');
const ch = await conn.createChannel();

// 1. Direct Exchange：精确匹配 Routing Key
await ch.assertExchange('orders.direct', 'direct', { durable: true });
await ch.assertQueue('orders.payment');
await ch.bindQueue('orders.payment', 'orders.direct', 'payment');

// 2. Topic Exchange：模式匹配（支持 * 和 #）
await ch.assertExchange('logs.topic', 'topic', { durable: true });
await ch.assertQueue('logs.error');
await ch.bindQueue('logs.error', 'logs.topic', 'kernel.error.*');

// 3. Fanout Exchange：广播到所有绑定队列
await ch.assertExchange('notifications.fanout', 'fanout', { durable: true });

// 4. Headers Exchange：根据消息头属性匹配
await ch.assertExchange('events.headers', 'headers', { durable: true });

// producer.js
const amqp = require('amqplib');

async function sendOrder(order) {
    const conn = await amqp.connect(process.env.RABBITMQ_URL);
    const ch = await conn.createChannel();

    // 声明 durable Exchange 和 Queue
    await ch.assertExchange('orders.topic', 'topic', { durable: true });
    await ch.assertQueue('orders.processing', {
        durable: true,
        arguments: {
            'x-dead-letter-exchange': 'orders.dlx',
            'x-dead-letter-routing-key': 'orders.failed'
        }
    });
    await ch.bindQueue('orders.processing', 'orders.topic', 'order.created');

    // 发送消息（persistent 确保持久化到磁盘）
    const msg = Buffer.from(JSON.stringify(order));
    ch.publish('orders.topic', 'order.created', msg, {
        persistent: true,
        messageId: order.id,
        timestamp: Date.now()
    });

    console.log(`[Producer] Order ${order.id} sent`);
    await ch.close();
    await conn.close();
}

// consumer.js
async function consumeOrders() {
    const conn = await amqp.connect(process.env.RABBITMQ_URL);
    const ch = await conn.createChannel();

    // 每次只接收 1 条，处理完再取下一条
    await ch.prefetch(1);

    await ch.consume('orders.processing', async (msg) => {
        if (!msg) return;

        try {
            const order = JSON.parse(msg.content.toString());
            console.log(`[Consumer] Processing order ${order.id}`);

            // 模拟业务处理
            await processOrder(order);

            // 确认消息已处理
            ch.ack(msg);
        } catch (err) {
            console.error(`[Consumer] Failed:`, err.message);

            // 拒绝消息，requeue=false 则进入死信队列
            ch.nack(msg, false, false);
        }
    });
}

async function processOrder(order) {
    // 订单处理逻辑
    await new Promise(resolve => setTimeout(resolve, 100));
}

// 声明死信交换器和队列
await ch.assertExchange('orders.dlx', 'topic', { durable: true });
await ch.assertQueue('orders.dead', { durable: true });
await ch.bindQueue('orders.dead', 'orders.dlx', 'orders.failed');

// 主队列绑定死信参数（在 assertQueue 时声明）
await ch.assertQueue('orders.processing', {
    durable: true,
    arguments: {
        'x-message-ttl': 300000,           // 消息 5 分钟过期
        'x-dead-letter-exchange': 'orders.dlx',
        'x-dead-letter-routing-key': 'orders.failed',
        'x-max-retries': 3                 // 最大重试次数（配合插件或代码实现）
    }
});

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
    clientId: 'order-service',
    brokers: ['kafka1:9092', 'kafka2:9092'],
    retry: {
        initialRetryTime: 100,
        retries: 8
    }
});

// ========== 生产者 ==========
const producer = kafka.producer({
    idempotent: true,          // 幂等生产者，防止重复发送
    transactionalId: 'order-producer'
});

async function sendOrderEvent(order) {
    await producer.connect();

    // 发送带 Key 的消息（相同 Key 进入同一 Partition，保证顺序）
    await producer.send({
        topic: 'orders',
        messages: [{
            key: order.userId,              // 按用户 ID 分区
            value: JSON.stringify(order),
            headers: {
                'event-type': 'order.created',
                'version': '1.0'
            }
        }]
    });

    await producer.disconnect();
}

// ========== 消费者 ==========
const consumer = kafka.consumer({
    groupId: 'order-processor-group',
    sessionTimeout: 30000,
    heartbeatInterval: 3000
});

async function consumeOrderEvents() {
    await consumer.connect();
    await consumer.subscribe({ topic: 'orders', fromBeginning: false });

    await consumer.run({
        autoCommit: false,          // 手动提交偏移量
        eachBatchAutoResolve: false,
        eachBatch: async ({ batch, resolveOffset, heartbeat, commitOffsetsIfNecessary }) => {
            for (const message of batch.messages) {
                try {
                    const order = JSON.parse(message.value.toString());
                    console.log(`[Kafka] Processing order: ${order.id}, partition: ${batch.partition}, offset: ${message.offset}`);

                    await processOrder(order);

                    // 处理成功，提交偏移量
                    await resolveOffset(message.offset);
                    await heartbeat();
                } catch (err) {
                    console.error(`[Kafka] Processing failed:`, err);
                    // 不提交偏移量，下次重试
                    throw err;
                }
            }
            await commitOffsetsIfNecessary();
        }
    });
}

// 监听重平衡事件
consumer.on(consumer.events.GROUP_JOIN, (event) => {
    console.log(`Joined group: ${event.payload.groupId}, generation: ${event.payload.generationId}`);
});

consumer.on(consumer.events.REBALANCING, (event) => {
    console.log('Consumer group rebalancing...');
});

const { Queue, Worker } = require('bullmq');
const Redis = require('ioredis');

const connection = new Redis({ host: 'localhost', port: 6379, maxRetriesPerRequest: null });

// 创建队列
const emailQueue = new Queue('email', { connection });

// 添加任务
async function enqueueEmail(data) {
    const job = await emailQueue.add('send-email', data, {
        attempts: 3,                    // 失败重试 3 次
        backoff: {
            type: 'exponential',        // 指数退避
            delay: 5000                 // 初始延迟 5 秒
        },
        removeOnComplete: 100,          // 保留最近 100 条完成记录
        removeOnFail: 50                // 保留最近 50 条失败记录
    });
    console.log(`Job ${job.id} enqueued`);
}

// 创建 Worker 处理任务
const emailWorker = new Worker('email', async (job) => {
    console.log(`[Worker] Processing job ${job.id}:`, job.data);

    // 模拟发送邮件
    await sendEmail(job.data);

    return { sent: true, timestamp: Date.now() };
}, {
    connection,
    concurrency: 5,                     // 并发处理 5 个任务
    limiter: {
        max: 100,                       // 每秒最多 100 个任务
        duration: 1000
    }
});

emailWorker.on('completed', (job, result) => {
    console.log(`Job ${job.id} completed:`, result);
});

emailWorker.on('failed', (job, err) => {
    console.error(`Job ${job.id} failed:`, err.message);
});

// ========== 延迟任务 ==========
// 10 秒后执行
await emailQueue.add('scheduled-email', { to: 'user@example.com' }, {
    delay: 10000
});

// 指定确切时间执行
const delay = new Date('2026-08-18T09:00:00+08:00').getTime() - Date.now();
await emailQueue.add('morning-digest', { userId: 123 }, { delay });

// ========== 重复任务（Cron 风格）==========
const { QueueScheduler } = require('bullmq');
new QueueScheduler('reports', { connection });

await emailQueue.add('daily-report', { type: 'analytics' }, {
    repeat: {
        cron: '0 9 * * *',              // 每天上午 9 点
        tz: 'Asia/Shanghai'
    }
});

await emailQueue.add('weekly-summary', {}, {
    repeat: {
        every: 7 * 24 * 60 * 60 * 1000, // 每 7 天
        limit: 52                       // 最多执行 52 次
    }
});

// 监听队列事件
emailQueue.on('waiting', (jobId) => console.log(`Job ${jobId} is waiting`));
emailQueue.on('active', (job) => console.log(`Job ${job.id} is active`));
emailQueue.on('stalled', (jobId) => console.warn(`Job ${jobId} stalled`));

// 获取队列状态
async function getQueueStatus() {
    const [waiting, active, completed, failed, delayed] = await Promise.all([
        emailQueue.getWaitingCount(),
        emailQueue.getActiveCount(),
        emailQueue.getCompletedCount(),
        emailQueue.getFailedCount(),
        emailQueue.getDelayedCount()
    ]);

    return { waiting, active, completed, failed, delayed };
}

const { SQSClient, SendMessageCommand, ReceiveMessageCommand, DeleteMessageCommand } = require('@aws-sdk/client-sqs');

const sqs = new SQSClient({ region: 'ap-southeast-1' });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

// 发送消息
async function sendSQSMessage(payload) {
    await sqs.send(new SendMessageCommand({
        QueueUrl: QUEUE_URL,
        MessageBody: JSON.stringify(payload),
        MessageAttributes: {
            'EventType': { StringValue: 'OrderCreated', DataType: 'String' }
        }
    }));
}

// 接收并处理消息
async function pollSQS() {
    const result = await sqs.send(new ReceiveMessageCommand({
        QueueUrl: QUEUE_URL,
        MaxNumberOfMessages: 10,
        WaitTimeSeconds: 20,            // 长轮询
        VisibilityTimeout: 300          // 处理超时 5 分钟
    }));

    for (const message of result.Messages || []) {
        try {
            const body = JSON.parse(message.Body);
            await processMessage(body);

            // 删除已处理消息
            await sqs.send(new DeleteMessageCommand({
                QueueUrl: QUEUE_URL,
                ReceiptHandle: message.ReceiptHandle
            }));
        } catch (err) {
            console.error('SQS processing failed:', err);
            // 不删除消息，VisibilityTimeout 到期后重试
        }
    }
}

const { SNSClient, PublishCommand } = require('@aws-sdk/client-sns');

const sns = new SNSClient({ region: 'ap-southeast-1' });
const TOPIC_ARN = process.env.SNS_TOPIC_ARN;

// SNS 发布消息
async function publishEvent(event) {
    await sns.send(new PublishCommand({
        TopicArn: TOPIC_ARN,
        Message: JSON.stringify(event),
        MessageAttributes: {
            'env': { DataType: 'String', StringValue: 'production' }
        }
    }));
}

// 事件存储
const events = [
    { type: 'OrderCreated', data: { orderId: '001', items: [...] } },
    { type: 'PaymentProcessed', data: { orderId: '001', amount: 199 } },
    { type: 'OrderShipped', data: { orderId: '001', tracking: 'SF123' } }
];

// 状态重建
function rebuildState(events) {
    return events.reduce((state, event) => {
        switch (event.type) {
            case 'OrderCreated':
                return { ...state, id: event.data.orderId, items: event.data.items, status: 'created' };
            case 'PaymentProcessed':
                return { ...state, status: 'paid', paidAt: Date.now() };
            case 'OrderShipped':
                return { ...state, status: 'shipped', tracking: event.data.tracking };
            default:
                return state;
        }
    }, {});
}

// order-service/saga.js
const { Kafka } = require('kafkajs');
const kafka = new Kafka({ clientId: 'order-saga', brokers: ['kafka:9092'] });
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: 'order-saga-group' });

// Saga 步骤定义
const sagaSteps = {
    'order.created': {
        next: 'inventory.reserve',
        compensate: 'order.cancel'
    },
    'inventory.reserved': {
        next: 'payment.charge',
        compensate: 'inventory.release'
    },
    'payment.success': {
        next: 'shipment.create',
        compensate: 'payment.refund'
    },
    'shipment.created': {
        next: null,  // Saga 完成
        compensate: 'shipment.cancel'
    }
};

async function startSaga(order) {
    await producer.connect();
    await producer.send({
        topic: 'saga-events',
        messages: [{
            key: order.id,
            value: JSON.stringify({
                type: 'order.created',
                payload: order,
                sagaId: order.id,
                step: 0
            })
        }]
    });
}

async function handleSagaEvent(event) {
    const { type, payload, sagaId } = event;
    const step = sagaSteps[type];

    if (!step) return;

    try {
        // 执行本地事务
        await executeLocalTransaction(type, payload);

        if (step.next) {
            // 触发下一步
            await producer.send({
                topic: 'saga-events',
                messages: [{
                    key: sagaId,
                    value: JSON.stringify({
                        type: step.next,
                        payload,
                        sagaId,
                        step: event.step + 1
                    })
                }]
            });
        }
    } catch (err) {
        // 触发补偿事务
        console.error(`Saga step ${type} failed, triggering compensation`);
        await triggerCompensation(sagaId, step.compensate, payload);
    }
}

async function triggerCompensation(sagaId, compensateAction, payload) {
    await producer.send({
        topic: 'saga-compensation',
        messages: [{
            key: sagaId,
            value: JSON.stringify({ type: compensateAction, payload, sagaId })
        }]
    });
}

// saga-orchestrator.js
class OrderSagaOrchestrator {
    constructor() {
        this.state = new Map(); // sagaId -> currentStep
    }

    async execute(orderId, orderData) {
        const saga = { id: orderId, status: 'pending', steps: [] };

        try {
            // Step 1: 创建订单
            await this.callService('order-service', 'create', orderData);
            saga.steps.push({ service: 'order', action: 'create', status: 'ok' });

            // Step 2: 预留库存
            await this.callService('inventory-service', 'reserve', { orderId, items: orderData.items });
            saga.steps.push({ service: 'inventory', action: 'reserve', status: 'ok' });

            // Step 3: 扣款
            await this.callService('payment-service', 'charge', { orderId, amount: orderData.total });
            saga.steps.push({ service: 'payment', action: 'charge', status: 'ok' });

            // Step 4: 创建物流
            await this.callService('shipment-service', 'create', { orderId, address: orderData.address });
            saga.status = 'completed';

        } catch (err) {
            saga.status = 'failed';
            // 倒序执行补偿
            for (const step of [...saga.steps].reverse()) {
                await this.compensate(step, orderId);
            }
        }

        return saga;
    }

    async compensate(step, orderId) {
        const compensations = {
            'order:create': () => this.callService('order-service', 'cancel', { orderId }),
            'inventory:reserve': () => this.callService('inventory-service', 'release', { orderId }),
            'payment:charge': () => this.callService('payment-service', 'refund', { orderId })
        };

        const key = `${step.service}:${step.action}`;
        if (compensations[key]) {
            await compensations[key]();
        }
    }
}

// Kafka 按 Key 分区保证用户级顺序
await producer.send({
    topic: 'user-events',
    messages: userEvents.map(e => ({
        key: e.userId,      // 同一用户的事件进入同一分区
        value: JSON.stringify(e)
    }))
});

// RabbitMQ 单队列保证顺序（取消并发消费）
await ch.prefetch(1);       // 一次只处理一条

// 幂等消费者设计
const processedIds = new Set();     // 生产环境使用 Redis/DB

async function processMessage(msg) {
    const id = msg.messageId || msg.id;

    // 检查是否已处理
    if (await isProcessed(id)) {
        console.log(`Message ${id} already processed, skipping`);
        return;
    }

    // 业务处理 + 记录处理状态应在同一事务中
    await withTransaction(async (trx) => {
        await executeBusinessLogic(msg, trx);
        await markAsProcessed(id, trx);
    });
}

// Kafka 事务型 Exactly-Once
const transaction = await producer.transaction();
try {
    await transaction.send({ topic: 'output', messages: [...] });
    await transaction.sendOffsets({
        consumerGroupId: 'my-group',
        topics: [{
            topic: 'input',
            partitions: [{ partition: 0, offset: '10' }]
        }]
    });
    await transaction.commit();
} catch (err) {
    await transaction.abort();
}

// 使用 rabbitmq-management HTTP API
const axios = require('axios');

async function getRabbitMQMetrics() {
    const res = await axios.get('http://localhost:15672/api/queues', {
        auth: { username: 'guest', password: 'guest' }
    });

    return res.data.map(q => ({
        name: q.name,
        messages_ready: q.messages_ready,       // 待消费消息
        messages_unacknowledged: q.messages_unacknowledged, // 已投递未确认
        consumers: q.consumers,
        message_stats: q.message_stats
    }));
}

const { Kafka } = require('kafkajs');
const kafka = new Kafka({ clientId: 'monitor', brokers: ['kafka:9092'] });
const admin = kafka.admin();

async function getConsumerLag(groupId) {
    await admin.connect();
    const lag = await admin.fetchOffsets({ groupId, topic: 'orders' });
    const offsets = await admin.fetchTopicOffsets('orders');

    const result = lag.map(({ partition, offset }) => {
        const latest = offsets.find(o => o.partition === partition);
        return {
            partition,
            consumerOffset: parseInt(offset),
            logEndOffset: latest.offset,
            lag: latest.offset - parseInt(offset)
        };
    });

    await admin.disconnect();
    return result;
}

const { Queue } = require('bullmq');
const queue = new Queue('email', { connection });

async function getBullMetrics() {
    const [waiting, active, completed, failed, delayed, paused] = await Promise.all([
        queue.getWaitingCount(),
        queue.getActiveCount(),
        queue.getCompletedCount(),
        queue.getFailedCount(),
        queue.getDelayedCount(),
        queue.getPausedCount()
    ]);

    const metrics = {
        waiting, active, completed, failed, delayed, paused,
        total: waiting + active + completed + failed + delayed
    };

    // 推送到 Prometheus
    metricsQueueDepth.set({ queue: 'email' }, waiting);
    metricsActiveJobs.set({ queue: 'email' }, active);
    metricsFailedJobs.inc({ queue: 'email' }, failed);

    return metrics;
}

// Express 健康检查端点
app.get('/health/queue', async (req, res) => {
    const statuses = await Promise.all([
        checkRabbitMQConnection(),
        checkKafkaConnection(),
        checkRedisConnection()
    ]);

    const allHealthy = statuses.every(s => s.healthy);
    res.status(allHealthy ? 200 : 503).json({
        status: allHealthy ? 'healthy' : 'unhealthy',
        services: statuses
    });
});