消息队列实战指南

1. 消息队列概述

什么是消息队列

消息队列（Message Queue，MQ）是一种应用程序间的通信方法，通过在消息的传输过程中保存消息的容器来实现。消息队列提供了异步的通信协议，每一个贮列中的记录包含详细说明的数据，包含发生的时间，输入设备的种类，以及特定的输入参数。

核心概念

• 生产者（Producer）：发送消息的应用程序
• 消费者（Consumer）：接收和处理消息的应用程序
• 队列（Queue）：存储消息的容器
• 交换机（Exchange）：路由消息到队列的组件
• 路由键（Routing Key）：消息路由的标识
• 绑定（Binding）：交换机和队列之间的关联

消息队列的优势

1. 解耦：生产者和消费者不需要直接交互
2. 异步处理：提高系统响应速度
3. 削峰填谷：处理突发流量
4. 可靠性：消息持久化和确认机制
5. 扩展性：支持水平扩展

2. RabbitMQ实战

基础配置和连接

// rabbitmq-client.js - RabbitMQ客户端封装
const amqp = require('amqplib');
const EventEmitter = require('events');

class RabbitMQClient extends EventEmitter {
    constructor(config = {}) {
        super();
        this.config = {
            url: config.url || 'amqp://localhost',
            reconnectInterval: config.reconnectInterval || 5000,
            maxReconnectAttempts: config.maxReconnectAttempts || 10,
            ...config
        };
        
        this.connection = null;
        this.channel = null;
        this.isConnected = false;
        this.reconnectAttempts = 0;
        
        this.connect();
    }

    async connect() {
        try {
            console.log('Connecting to RabbitMQ...');
            this.connection = await amqp.connect(this.config.url);
            this.channel = await this.connection.createChannel();
            
            this.isConnected = true;
            this.reconnectAttempts = 0;
            
            console.log('Connected to RabbitMQ successfully');
            this.emit('connected');

            // 监听连接错误
            this.connection.on('error', (err) => {
                console.error('RabbitMQ connection error:', err);
                this.isConnected = false;
                this.emit('error', err);
            });

            this.connection.on('close', () => {
                console.log('RabbitMQ connection closed');
                this.isConnected = false;
                this.emit('disconnected');
                this.reconnect();
            });

        } catch (error) {
            console.error('Failed to connect to RabbitMQ:', error);
            this.isConnected = false;
            this.emit('error', error);
            this.reconnect();
        }
    }

    async reconnect() {
        if (this.reconnectAttempts >= this.config.maxReconnectAttempts) {
            console.error('Max reconnection attempts reached');
            this.emit('maxReconnectAttemptsReached');
            return;
        }

        this.reconnectAttempts++;
        console.log(`Reconnecting to RabbitMQ (attempt ${this.reconnectAttempts})...`);
        
        setTimeout(() => {
            this.connect();
        }, this.config.reconnectInterval);
    }

    async assertQueue(queueName, options = {}) {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        const defaultOptions = {
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: {}
        };

        return await this.channel.assertQueue(queueName, { ...defaultOptions, ...options });
    }

    async assertExchange(exchangeName, type = 'direct', options = {}) {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        const defaultOptions = {
            durable: true,
            internal: false,
            autoDelete: false,
            arguments: {}
        };

        return await this.channel.assertExchange(exchangeName, type, { ...defaultOptions, ...options });
    }

    async bindQueue(queueName, exchangeName, routingKey = '') {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        return await this.channel.bindQueue(queueName, exchangeName, routingKey);
    }

    async publish(exchangeName, routingKey, message, options = {}) {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        const messageBuffer = Buffer.from(JSON.stringify(message));
        const defaultOptions = {
            persistent: true,
            timestamp: Date.now(),
            messageId: this.generateMessageId()
        };

        return this.channel.publish(
            exchangeName, 
            routingKey, 
            messageBuffer, 
            { ...defaultOptions, ...options }
        );
    }

    async sendToQueue(queueName, message, options = {}) {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        const messageBuffer = Buffer.from(JSON.stringify(message));
        const defaultOptions = {
            persistent: true,
            timestamp: Date.now(),
            messageId: this.generateMessageId()
        };

        return this.channel.sendToQueue(queueName, messageBuffer, { ...defaultOptions, ...options });
    }

    async consume(queueName, callback, options = {}) {
        if (!this.isConnected) {
            throw new Error('Not connected to RabbitMQ');
        }

        const defaultOptions = {
            noAck: false,
            exclusive: false,
            priority: 0,
            consumerTag: ''
        };

        return await this.channel.consume(queueName, async (msg) => {
            if (msg) {
                try {
                    const content = JSON.parse(msg.content.toString());
                    const result = await callback(content, msg);
                    
                    if (!options.noAck && result !== false) {
                        this.channel.ack(msg);
                    }
                } catch (error) {
                    console.error('Error processing message:', error);
                    
                    // 重试机制
                    if (msg.properties.headers && msg.properties.headers['x-retry-count']) {
                        const retryCount = msg.properties.headers['x-retry-count'];
                        if (retryCount < 3) {
                            // 重新发送到队列
                            setTimeout(() => {
                                this.sendToQueue(queueName, JSON.parse(msg.content.toString()), {
                                    headers: { 'x-retry-count': retryCount + 1 }
                                });
                            }, 1000 * Math.pow(2, retryCount)); // 指数退避
                        } else {
                            // 发送到死信队列
                            this.sendToQueue(`${queueName}.dlq`, JSON.parse(msg.content.toString()));
                        }
                    } else {
                        this.sendToQueue(queueName, JSON.parse(msg.content.toString()), {
                            headers: { 'x-retry-count': 1 }
                        });
                    }
                    
                    this.channel.nack(msg, false, false);
                }
            }
        }, { ...defaultOptions, ...options });
    }

    async close() {
        if (this.channel) {
            await this.channel.close();
        }
        if (this.connection) {
            await this.connection.close();
        }
        this.isConnected = false;
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = RabbitMQClient;

工作队列模式

// work-queue.js - 工作队列实现
const RabbitMQClient = require('./rabbitmq-client');

class WorkQueue {
    constructor(queueName, rabbitMQConfig = {}) {
        this.queueName = queueName;
        this.client = new RabbitMQClient(rabbitMQConfig);
        this.isReady = false;
        
        this.client.on('connected', async () => {
            await this.setup();
        });
    }

    async setup() {
        try {
            // 声明队列
            await this.client.assertQueue(this.queueName, {
                durable: true,
                arguments: {
                    'x-max-priority': 10, // 支持优先级
                    'x-message-ttl': 300000, // 消息TTL 5分钟
                    'x-dead-letter-exchange': `${this.queueName}.dlx`,
                    'x-dead-letter-routing-key': `${this.queueName}.dlq`
                }
            });

            // 声明死信交换机和队列
            await this.client.assertExchange(`${this.queueName}.dlx`, 'direct');
            await this.client.assertQueue(`${this.queueName}.dlq`, { durable: true });
            await this.client.bindQueue(`${this.queueName}.dlq`, `${this.queueName}.dlx`, `${this.queueName}.dlq`);

            this.isReady = true;
            console.log(`Work queue ${this.queueName} is ready`);
        } catch (error) {
            console.error('Failed to setup work queue:', error);
        }
    }

    async addJob(jobData, options = {}) {
        if (!this.isReady) {
            throw new Error('Work queue is not ready');
        }

        const job = {
            id: this.generateJobId(),
            data: jobData,
            createdAt: new Date().toISOString(),
            attempts: 0,
            maxAttempts: options.maxAttempts || 3,
            delay: options.delay || 0
        };

        const messageOptions = {
            priority: options.priority || 0,
            expiration: options.ttl || 300000,
            headers: {
                'job-type': options.jobType || 'default',
                'created-by': options.createdBy || 'system'
            }
        };

        if (job.delay > 0) {
            // 延迟任务
            messageOptions.headers['x-delay'] = job.delay;
            await this.client.publish(`${this.queueName}.delayed`, '', job, messageOptions);
        } else {
            await this.client.sendToQueue(this.queueName, job, messageOptions);
        }

        console.log(`Job ${job.id} added to queue ${this.queueName}`);
        return job.id;
    }

    async processJobs(processor, options = {}) {
        if (!this.isReady) {
            throw new Error('Work queue is not ready');
        }

        const concurrency = options.concurrency || 1;
        
        // 设置预取数量
        await this.client.channel.prefetch(concurrency);

        return await this.client.consume(this.queueName, async (job, msg) => {
            const startTime = Date.now();
            
            try {
                console.log(`Processing job ${job.id}`);
                
                // 执行任务处理器
                const result = await processor(job.data, job);
                
                const processingTime = Date.now() - startTime;
                console.log(`Job ${job.id} completed in ${processingTime}ms`);
                
                // 记录成功指标
                this.recordJobMetrics(job, 'completed', processingTime);
                
                return result;
            } catch (error) {
                console.error(`Job ${job.id} failed:`, error);
                
                job.attempts++;
                job.lastError = error.message;
                job.lastAttemptAt = new Date().toISOString();

                if (job.attempts < job.maxAttempts) {
                    // 重试任务
                    const delay = Math.pow(2, job.attempts) * 1000; // 指数退避
                    console.log(`Retrying job ${job.id} in ${delay}ms (attempt ${job.attempts})`);
                    
                    setTimeout(async () => {
                        await this.client.sendToQueue(this.queueName, job);
                    }, delay);
                } else {
                    // 发送到死信队列
                    console.log(`Job ${job.id} failed permanently, sending to DLQ`);
                    await this.client.sendToQueue(`${this.queueName}.dlq`, job);
                }

                const processingTime = Date.now() - startTime;
                this.recordJobMetrics(job, 'failed', processingTime, error);
                
                throw error;
            }
        }, { noAck: false });
    }

    recordJobMetrics(job, status, processingTime, error = null) {
        // 这里可以集成监控系统，如Prometheus
        const metrics = {
            jobId: job.id,
            queueName: this.queueName,
            status,
            processingTime,
            attempts: job.attempts,
            timestamp: new Date().toISOString()
        };

        if (error) {
            metrics.error = error.message;
        }

        console.log('Job metrics:', metrics);
        // 发送到监控系统
        // prometheus.recordJobMetrics(metrics);
    }

    generateJobId() {
        return `job_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }

    async getQueueStats() {
        if (!this.isReady) {
            throw new Error('Work queue is not ready');
        }

        const queueInfo = await this.client.channel.checkQueue(this.queueName);
        return {
            messageCount: queueInfo.messageCount,
            consumerCount: queueInfo.consumerCount
        };
    }

    async purgeQueue() {
        if (!this.isReady) {
            throw new Error('Work queue is not ready');
        }

        return await this.client.channel.purgeQueue(this.queueName);
    }
}

module.exports = WorkQueue;

发布/订阅模式

// pub-sub.js - 发布订阅实现
const RabbitMQClient = require('./rabbitmq-client');

class PubSub {
    constructor(exchangeName, rabbitMQConfig = {}) {
        this.exchangeName = exchangeName;
        this.client = new RabbitMQClient(rabbitMQConfig);
        this.isReady = false;
        
        this.client.on('connected', async () => {
            await this.setup();
        });
    }

    async setup() {
        try {
            // 声明fanout交换机
            await this.client.assertExchange(this.exchangeName, 'fanout', {
                durable: true
            });

            this.isReady = true;
            console.log(`PubSub exchange ${this.exchangeName} is ready`);
        } catch (error) {
            console.error('Failed to setup PubSub:', error);
        }
    }

    async publish(message, options = {}) {
        if (!this.isReady) {
            throw new Error('PubSub is not ready');
        }

        const messageData = {
            id: this.generateMessageId(),
            data: message,
            timestamp: new Date().toISOString(),
            publisher: options.publisher || 'anonymous'
        };

        const messageOptions = {
            persistent: true,
            headers: {
                'message-type': options.messageType || 'broadcast',
                'publisher': messageData.publisher
            }
        };

        await this.client.publish(this.exchangeName, '', messageData, messageOptions);
        console.log(`Message ${messageData.id} published to ${this.exchangeName}`);
        
        return messageData.id;
    }

    async subscribe(subscriber, callback, options = {}) {
        if (!this.isReady) {
            throw new Error('PubSub is not ready');
        }

        // 为每个订阅者创建独立的队列
        const queueName = `${this.exchangeName}.${subscriber}`;
        
        await this.client.assertQueue(queueName, {
            exclusive: false,
            autoDelete: options.autoDelete || false,
            durable: true
        });

        await this.client.bindQueue(queueName, this.exchangeName, '');

        console.log(`Subscriber ${subscriber} subscribed to ${this.exchangeName}`);

        return await this.client.consume(queueName, async (message, msg) => {
            try {
                console.log(`Subscriber ${subscriber} received message ${message.id}`);
                await callback(message.data, message);
            } catch (error) {
                console.error(`Subscriber ${subscriber} failed to process message:`, error);
                throw error;
            }
        }, { noAck: false });
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = PubSub;

路由模式

// routing.js - 路由模式实现
const RabbitMQClient = require('./rabbitmq-client');

class Routing {
    constructor(exchangeName, rabbitMQConfig = {}) {
        this.exchangeName = exchangeName;
        this.client = new RabbitMQClient(rabbitMQConfig);
        this.isReady = false;
        
        this.client.on('connected', async () => {
            await this.setup();
        });
    }

    async setup() {
        try {
            // 声明direct交换机
            await this.client.assertExchange(this.exchangeName, 'direct', {
                durable: true
            });

            this.isReady = true;
            console.log(`Routing exchange ${this.exchangeName} is ready`);
        } catch (error) {
            console.error('Failed to setup Routing:', error);
        }
    }

    async publish(routingKey, message, options = {}) {
        if (!this.isReady) {
            throw new Error('Routing is not ready');
        }

        const messageData = {
            id: this.generateMessageId(),
            routingKey,
            data: message,
            timestamp: new Date().toISOString(),
            publisher: options.publisher || 'anonymous'
        };

        const messageOptions = {
            persistent: true,
            headers: {
                'routing-key': routingKey,
                'message-type': options.messageType || 'routed',
                'publisher': messageData.publisher
            }
        };

        await this.client.publish(this.exchangeName, routingKey, messageData, messageOptions);
        console.log(`Message ${messageData.id} published with routing key ${routingKey}`);
        
        return messageData.id;
    }

    async subscribe(routingKeys, subscriber, callback, options = {}) {
        if (!this.isReady) {
            throw new Error('Routing is not ready');
        }

        const queueName = `${this.exchangeName}.${subscriber}`;
        
        await this.client.assertQueue(queueName, {
            exclusive: false,
            autoDelete: options.autoDelete || false,
            durable: true
        });

        // 绑定多个路由键
        const keys = Array.isArray(routingKeys) ? routingKeys : [routingKeys];
        for (const routingKey of keys) {
            await this.client.bindQueue(queueName, this.exchangeName, routingKey);
            console.log(`Bound queue ${queueName} to routing key ${routingKey}`);
        }

        console.log(`Subscriber ${subscriber} subscribed to routing keys: ${keys.join(', ')}`);

        return await this.client.consume(queueName, async (message, msg) => {
            try {
                console.log(`Subscriber ${subscriber} received message ${message.id} with routing key ${message.routingKey}`);
                await callback(message.data, message);
            } catch (error) {
                console.error(`Subscriber ${subscriber} failed to process message:`, error);
                throw error;
            }
        }, { noAck: false });
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = Routing;

3. Apache Kafka实战

Kafka生产者

// kafka-producer.js - Kafka生产者实现
const { Kafka, Partitioners } = require('kafkajs');

class KafkaProducer {
    constructor(config = {}) {
        this.config = {
            clientId: config.clientId || 'my-app-producer',
            brokers: config.brokers || ['localhost:9092'],
            ...config
        };

        this.kafka = new Kafka(this.config);
        this.producer = this.kafka.producer({
            createPartitioner: Partitioners.LegacyPartitioner,
            maxInFlightRequests: 1,
            idempotent: true,
            transactionTimeout: 30000,
            retry: {
                initialRetryTime: 100,
                retries: 8
            }
        });

        this.isConnected = false;
    }

    async connect() {
        try {
            await this.producer.connect();
            this.isConnected = true;
            console.log('Kafka producer connected');
        } catch (error) {
            console.error('Failed to connect Kafka producer:', error);
            throw error;
        }
    }

    async send(topic, messages, options = {}) {
        if (!this.isConnected) {
            throw new Error('Producer is not connected');
        }

        // 格式化消息
        const formattedMessages = Array.isArray(messages) ? messages : [messages];
        const kafkaMessages = formattedMessages.map(msg => ({
            key: msg.key || null,
            value: JSON.stringify({
                id: this.generateMessageId(),
                data: msg.value || msg,
                timestamp: Date.now(),
                producer: this.config.clientId
            }),
            partition: msg.partition,
            headers: {
                'content-type': 'application/json',
                'producer-id': this.config.clientId,
                ...msg.headers
            }
        }));

        try {
            const result = await this.producer.send({
                topic,
                messages: kafkaMessages,
                acks: options.acks || -1, // 等待所有副本确认
                timeout: options.timeout || 30000,
                compression: options.compression || 'gzip'
            });

            console.log(`Sent ${kafkaMessages.length} messages to topic ${topic}`);
            return result;
        } catch (error) {
            console.error('Failed to send messages:', error);
            throw error;
        }
    }

    async sendBatch(topicMessages, options = {}) {
        if (!this.isConnected) {
            throw new Error('Producer is not connected');
        }

        const batch = topicMessages.map(({ topic, messages }) => ({
            topic,
            messages: (Array.isArray(messages) ? messages : [messages]).map(msg => ({
                key: msg.key || null,
                value: JSON.stringify({
                    id: this.generateMessageId(),
                    data: msg.value || msg,
                    timestamp: Date.now(),
                    producer: this.config.clientId
                }),
                partition: msg.partition,
                headers: {
                    'content-type': 'application/json',
                    'producer-id': this.config.clientId,
                    ...msg.headers
                }
            }))
        }));

        try {
            const result = await this.producer.sendBatch({
                topicMessages: batch,
                acks: options.acks || -1,
                timeout: options.timeout || 30000,
                compression: options.compression || 'gzip'
            });

            console.log(`Sent batch messages to ${batch.length} topics`);
            return result;
        } catch (error) {
            console.error('Failed to send batch messages:', error);
            throw error;
        }
    }

    // 事务支持
    async sendTransaction(transactionMessages) {
        if (!this.isConnected) {
            throw new Error('Producer is not connected');
        }

        const transaction = await this.producer.transaction();

        try {
            for (const { topic, messages } of transactionMessages) {
                await transaction.send({
                    topic,
                    messages: (Array.isArray(messages) ? messages : [messages]).map(msg => ({
                        key: msg.key || null,
                        value: JSON.stringify({
                            id: this.generateMessageId(),
                            data: msg.value || msg,
                            timestamp: Date.now(),
                            producer: this.config.clientId
                        }),
                        headers: {
                            'content-type': 'application/json',
                            'producer-id': this.config.clientId,
                            'transaction-id': transaction.transactionId,
                            ...msg.headers
                        }
                    }))
                });
            }

            await transaction.commit();
            console.log('Transaction committed successfully');
        } catch (error) {
            await transaction.abort();
            console.error('Transaction aborted:', error);
            throw error;
        }
    }

    async disconnect() {
        if (this.isConnected) {
            await this.producer.disconnect();
            this.isConnected = false;
            console.log('Kafka producer disconnected');
        }
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = KafkaProducer;

Kafka消费者

// kafka-consumer.js - Kafka消费者实现
const { Kafka } = require('kafkajs');

class KafkaConsumer {
    constructor(config = {}) {
        this.config = {
            clientId: config.clientId || 'my-app-consumer',
            brokers: config.brokers || ['localhost:9092'],
            groupId: config.groupId || 'my-consumer-group',
            ...config
        };

        this.kafka = new Kafka(this.config);
        this.consumer = this.kafka.consumer({
            groupId: this.config.groupId,
            sessionTimeout: 30000,
            rebalanceTimeout: 60000,
            heartbeatInterval: 3000,
            maxBytesPerPartition: 1048576,
            minBytes: 1,
            maxBytes: 10485760,
            maxWaitTimeInMs: 5000,
            retry: {
                initialRetryTime: 100,
                retries: 8
            }
        });

        this.isConnected = false;
        this.messageHandlers = new Map();
    }

    async connect() {
        try {
            await this.consumer.connect();
            this.isConnected = true;
            console.log('Kafka consumer connected');
        } catch (error) {
            console.error('Failed to connect Kafka consumer:', error);
            throw error;
        }
    }

    async subscribe(topics, handler, options = {}) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        const topicList = Array.isArray(topics) ? topics : [topics];
        
        // 订阅主题
        await this.consumer.subscribe({
            topics: topicList,
            fromBeginning: options.fromBeginning || false
        });

        // 注册消息处理器
        for (const topic of topicList) {
            this.messageHandlers.set(topic, handler);
        }

        console.log(`Subscribed to topics: ${topicList.join(', ')}`);
    }

    async run(options = {}) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        await this.consumer.run({
            autoCommit: options.autoCommit !== false,
            autoCommitInterval: options.autoCommitInterval || 5000,
            autoCommitThreshold: options.autoCommitThreshold || 100,
            eachMessage: async ({ topic, partition, message }) => {
                const startTime = Date.now();
                
                try {
                    // 解析消息
                    const messageData = JSON.parse(message.value.toString());
                    const messageInfo = {
                        topic,
                        partition,
                        offset: message.offset,
                        key: message.key ? message.key.toString() : null,
                        headers: message.headers,
                        timestamp: message.timestamp,
                        ...messageData
                    };

                    console.log(`Processing message from topic ${topic}, partition ${partition}, offset ${message.offset}`);

                    // 获取对应的处理器
                    const handler = this.messageHandlers.get(topic);
                    if (handler) {
                        await handler(messageInfo.data, messageInfo);
                    } else {
                        console.warn(`No handler found for topic ${topic}`);
                    }

                    const processingTime = Date.now() - startTime;
                    this.recordMessageMetrics(messageInfo, 'processed', processingTime);

                } catch (error) {
                    const processingTime = Date.now() - startTime;
                    console.error(`Failed to process message from topic ${topic}:`, error);
                    
                    this.recordMessageMetrics({
                        topic,
                        partition,
                        offset: message.offset
                    }, 'failed', processingTime, error);

                    // 错误处理策略
                    if (options.errorHandler) {
                        await options.errorHandler(error, { topic, partition, message });
                    } else {
                        // 默认错误处理：记录错误但不中断处理
                        console.error('Message processing failed, continuing...');
                    }
                }
            },
            eachBatch: options.batchProcessor ? async ({ batch, resolveOffset, heartbeat, isRunning, isStale }) => {
                const startTime = Date.now();
                
                try {
                    const messages = batch.messages.map(message => {
                        const messageData = JSON.parse(message.value.toString());
                        return {
                            topic: batch.topic,
                            partition: batch.partition,
                            offset: message.offset,
                            key: message.key ? message.key.toString() : null,
                            headers: message.headers,
                            timestamp: message.timestamp,
                            ...messageData
                        };
                    });

                    console.log(`Processing batch of ${messages.length} messages from topic ${batch.topic}`);

                    await options.batchProcessor(messages, {
                        batch,
                        resolveOffset,
                        heartbeat,
                        isRunning,
                        isStale
                    });

                    const processingTime = Date.now() - startTime;
                    this.recordBatchMetrics(batch, 'processed', processingTime);

                } catch (error) {
                    const processingTime = Date.now() - startTime;
                    console.error(`Failed to process batch from topic ${batch.topic}:`, error);
                    
                    this.recordBatchMetrics(batch, 'failed', processingTime, error);
                    throw error;
                }
            } : undefined
        });
    }

    async commitOffsets(offsets) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        await this.consumer.commitOffsets(offsets);
        console.log('Offsets committed manually');
    }

    async seek(topicPartitions) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        for (const { topic, partition, offset } of topicPartitions) {
            this.consumer.seek({ topic, partition, offset });
            console.log(`Seeked to offset ${offset} for topic ${topic}, partition ${partition}`);
        }
    }

    async pause(topicPartitions) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        this.consumer.pause(topicPartitions);
        console.log('Consumer paused for specified topic partitions');
    }

    async resume(topicPartitions) {
        if (!this.isConnected) {
            throw new Error('Consumer is not connected');
        }

        this.consumer.resume(topicPartitions);
        console.log('Consumer resumed for specified topic partitions');
    }

    async disconnect() {
        if (this.isConnected) {
            await this.consumer.disconnect();
            this.isConnected = false;
            console.log('Kafka consumer disconnected');
        }
    }

    recordMessageMetrics(message, status, processingTime, error = null) {
        const metrics = {
            topic: message.topic,
            partition: message.partition,
            offset: message.offset,
            status,
            processingTime,
            timestamp: new Date().toISOString(),
            consumerGroup: this.config.groupId
        };

        if (error) {
            metrics.error = error.message;
        }

        console.log('Message metrics:', metrics);
        // 发送到监控系统
        // prometheus.recordMessageMetrics(metrics);
    }

    recordBatchMetrics(batch, status, processingTime, error = null) {
        const metrics = {
            topic: batch.topic,
            partition: batch.partition,
            batchSize: batch.messages.length,
            firstOffset: batch.firstOffset(),
            lastOffset: batch.lastOffset(),
            status,
            processingTime,
            timestamp: new Date().toISOString(),
            consumerGroup: this.config.groupId
        };

        if (error) {
            metrics.error = error.message;
        }

        console.log('Batch metrics:', metrics);
        // 发送到监控系统
        // prometheus.recordBatchMetrics(metrics);
    }
}

module.exports = KafkaConsumer;

Kafka管理工具

// kafka-admin.js - Kafka管理工具
const { Kafka } = require('kafkajs');

class KafkaAdmin {
    constructor(config = {}) {
        this.config = {
            clientId: config.clientId || 'my-app-admin',
            brokers: config.brokers || ['localhost:9092'],
            ...config
        };

        this.kafka = new Kafka(this.config);
        this.admin = this.kafka.admin();
        this.isConnected = false;
    }

    async connect() {
        try {
            await this.admin.connect();
            this.isConnected = true;
            console.log('Kafka admin connected');
        } catch (error) {
            console.error('Failed to connect Kafka admin:', error);
            throw error;
        }
    }

    async createTopics(topics) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        const topicConfigs = topics.map(topic => ({
            topic: topic.name,
            numPartitions: topic.numPartitions || 3,
            replicationFactor: topic.replicationFactor || 1,
            configEntries: [
                { name: 'cleanup.policy', value: topic.cleanupPolicy || 'delete' },
                { name: 'retention.ms', value: topic.retentionMs || '604800000' }, // 7天
                { name: 'segment.ms', value: topic.segmentMs || '86400000' }, // 1天
                { name: 'max.message.bytes', value: topic.maxMessageBytes || '1000000' },
                { name: 'compression.type', value: topic.compressionType || 'gzip' },
                ...(topic.configEntries || [])
            ]
        }));

        try {
            const result = await this.admin.createTopics({
                topics: topicConfigs,
                waitForLeaders: true,
                timeout: 30000
            });

            if (result) {
                console.log('Topics created successfully:', topics.map(t => t.name));
            } else {
                console.log('Topics already exist or creation failed');
            }

            return result;
        } catch (error) {
            console.error('Failed to create topics:', error);
            throw error;
        }
    }

    async deleteTopics(topicNames) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            await this.admin.deleteTopics({
                topics: topicNames,
                timeout: 30000
            });

            console.log('Topics deleted successfully:', topicNames);
        } catch (error) {
            console.error('Failed to delete topics:', error);
            throw error;
        }
    }

    async listTopics() {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            const topics = await this.admin.listTopics();
            console.log('Available topics:', topics);
            return topics;
        } catch (error) {
            console.error('Failed to list topics:', error);
            throw error;
        }
    }

    async describeTopics(topicNames) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            const metadata = await this.admin.fetchTopicMetadata({
                topics: topicNames
            });

            const topicDetails = metadata.topics.map(topic => ({
                name: topic.name,
                partitions: topic.partitions.map(partition => ({
                    partitionId: partition.partitionId,
                    leader: partition.leader,
                    replicas: partition.replicas,
                    isr: partition.isr
                }))
            }));

            console.log('Topic details:', topicDetails);
            return topicDetails;
        } catch (error) {
            console.error('Failed to describe topics:', error);
            throw error;
        }
    }

    async createPartitions(topicPartitions) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            await this.admin.createPartitions({
                topicPartitions,
                timeout: 30000
            });

            console.log('Partitions created successfully');
        } catch (error) {
            console.error('Failed to create partitions:', error);
            throw error;
        }
    }

    async alterConfigs(resources) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            await this.admin.alterConfigs({
                resources,
                timeout: 30000
            });

            console.log('Configurations altered successfully');
        } catch (error) {
            console.error('Failed to alter configurations:', error);
            throw error;
        }
    }

    async describeConfigs(resources) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            const result = await this.admin.describeConfigs({
                resources,
                includeSynonyms: true
            });

            console.log('Configuration details:', result);
            return result;
        } catch (error) {
            console.error('Failed to describe configurations:', error);
            throw error;
        }
    }

    async listGroups() {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            const groups = await this.admin.listGroups();
            console.log('Consumer groups:', groups);
            return groups;
        } catch (error) {
            console.error('Failed to list groups:', error);
            throw error;
        }
    }

    async describeGroups(groupIds) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            const result = await this.admin.describeGroups(groupIds);
            console.log('Group details:', result);
            return result;
        } catch (error) {
            console.error('Failed to describe groups:', error);
            throw error;
        }
    }

    async resetOffsets(groupId, topic, options = {}) {
        if (!this.isConnected) {
            throw new Error('Admin is not connected');
        }

        try {
            // 首先停止消费者组
            await this.admin.deleteGroups([groupId]);

            // 重置偏移量的逻辑需要根据具体需求实现
            // 这里提供一个基本框架
            console.log(`Reset offsets for group ${groupId} on topic ${topic}`);
            
            // 可以实现以下重置策略：
            // - earliest: 重置到最早的偏移量
            // - latest: 重置到最新的偏移量
            // - specific: 重置到指定的偏移量
            
        } catch (error) {
            console.error('Failed to reset offsets:', error);
            throw error;
        }
    }

    async disconnect() {
        if (this.isConnected) {
            await this.admin.disconnect();
            this.isConnected = false;
            console.log('Kafka admin disconnected');
        }
    }
}

module.exports = KafkaAdmin;

4. Redis消息队列实战

Redis List队列

// redis-list-queue.js - Redis List队列实现
const Redis = require('ioredis');

class RedisListQueue {
    constructor(queueName, redisConfig = {}) {
        this.queueName = queueName;
        this.redis = new Redis({
            host: redisConfig.host || 'localhost',
            port: redisConfig.port || 6379,
            password: redisConfig.password,
            db: redisConfig.db || 0,
            retryDelayOnFailover: 100,
            enableReadyCheck: false,
            maxRetriesPerRequest: null,
            ...redisConfig
        });

        this.processingQueue = `${queueName}:processing`;
        this.failedQueue = `${queueName}:failed`;
        this.delayedQueue = `${queueName}:delayed`;
        
        this.isProcessing = false;
    }

    async push(job, options = {}) {
        const jobData = {
            id: this.generateJobId(),
            data: job,
            createdAt: Date.now(),
            attempts: 0,
            maxAttempts: options.maxAttempts || 3,
            priority: options.priority || 0,
            delay: options.delay || 0
        };

        if (jobData.delay > 0) {
            // 延迟任务
            const executeAt = Date.now() + jobData.delay;
            await this.redis.zadd(this.delayedQueue, executeAt, JSON.stringify(jobData));
        } else if (options.priority > 0) {
            // 高优先级任务插入队列头部
            await this.redis.lpush(this.queueName, JSON.stringify(jobData));
        } else {
            // 普通任务插入队列尾部
            await this.redis.rpush(this.queueName, JSON.stringify(jobData));
        }

        console.log(`Job ${jobData.id} added to queue ${this.queueName}`);
        return jobData.id;
    }

    async pop(timeout = 0) {
        try {
            // 使用BLPOP实现阻塞式获取
            const result = await this.redis.blpop(this.queueName, timeout);
            if (result) {
                const [, jobStr] = result;
                const job = JSON.parse(jobStr);
                
                // 将任务移到处理队列
                await this.redis.lpush(this.processingQueue, jobStr);
                
                return job;
            }
            return null;
        } catch (error) {
            console.error('Failed to pop job:', error);
            throw error;
        }
    }

    async process(processor, options = {}) {
        if (this.isProcessing) {
            throw new Error('Queue is already being processed');
        }

        this.isProcessing = true;
        const concurrency = options.concurrency || 1;
        const workers = [];

        // 启动延迟任务处理器
        this.startDelayedJobProcessor();

        // 启动多个工作进程
        for (let i = 0; i < concurrency; i++) {
            workers.push(this.startWorker(processor, options));
        }

        console.log(`Started ${concurrency} workers for queue ${this.queueName}`);
        
        // 等待所有工作进程完成
        await Promise.all(workers);
    }

    async startWorker(processor, options) {
        while (this.isProcessing) {
            try {
                const job = await this.pop(5); // 5秒超时
                if (!job) continue;

                const startTime = Date.now();
                
                try {
                    console.log(`Processing job ${job.id}`);
                    
                    // 执行任务处理器
                    await processor(job.data, job);
                    
                    // 任务成功完成，从处理队列中移除
                    await this.removeFromProcessing(job);
                    
                    const processingTime = Date.now() - startTime;
                    console.log(`Job ${job.id} completed in ${processingTime}ms`);
                    
                } catch (error) {
                    console.error(`Job ${job.id} failed:`, error);
                    
                    job.attempts++;
                    job.lastError = error.message;
                    job.lastAttemptAt = Date.now();

                    // 从处理队列中移除
                    await this.removeFromProcessing(job);

                    if (job.attempts < job.maxAttempts) {
                        // 重试任务
                        const delay = Math.pow(2, job.attempts) * 1000; // 指数退避
                        job.delay = delay;
                        await this.push(job.data, job);
                        console.log(`Job ${job.id} will retry in ${delay}ms`);
                    } else {
                        // 移到失败队列
                        await this.redis.lpush(this.failedQueue, JSON.stringify(job));
                        console.log(`Job ${job.id} moved to failed queue`);
                    }
                }
            } catch (error) {
                console.error('Worker error:', error);
                // 短暂休息后继续
                await new Promise(resolve => setTimeout(resolve, 1000));
            }
        }
    }

    async startDelayedJobProcessor() {
        setInterval(async () => {
            try {
                const now = Date.now();
                // 获取到期的延迟任务
                const jobs = await this.redis.zrangebyscore(
                    this.delayedQueue, 
                    '-inf', 
                    now, 
                    'LIMIT', 0, 100
                );

                for (const jobStr of jobs) {
                    const job = JSON.parse(jobStr);
                    
                    // 移动到主队列
                    await this.redis.rpush(this.queueName, jobStr);
                    await this.redis.zrem(this.delayedQueue, jobStr);
                    
                    console.log(`Delayed job ${job.id} moved to main queue`);
                }
            } catch (error) {
                console.error('Delayed job processor error:', error);
            }
        }, 1000); // 每秒检查一次
    }

    async removeFromProcessing(job) {
        const jobStr = JSON.stringify(job);
        await this.redis.lrem(this.processingQueue, 1, jobStr);
    }

    async getStats() {
        const [waiting, processing, failed, delayed] = await Promise.all([
            this.redis.llen(this.queueName),
            this.redis.llen(this.processingQueue),
            this.redis.llen(this.failedQueue),
            this.redis.zcard(this.delayedQueue)
        ]);

        return {
            waiting,
            processing,
            failed,
            delayed,
            total: waiting + processing + failed + delayed
        };
    }

    async clear() {
        await Promise.all([
            this.redis.del(this.queueName),
            this.redis.del(this.processingQueue),
            this.redis.del(this.failedQueue),
            this.redis.del(this.delayedQueue)
        ]);
        console.log(`Queue ${this.queueName} cleared`);
    }

    async stop() {
        this.isProcessing = false;
        console.log(`Queue ${this.queueName} processing stopped`);
    }

    generateJobId() {
        return `job_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = RedisListQueue;

Redis Pub/Sub

// redis-pubsub.js - Redis发布订阅实现
const Redis = require('ioredis');
const EventEmitter = require('events');

class RedisPubSub extends EventEmitter {
    constructor(redisConfig = {}) {
        super();
        
        this.publisher = new Redis({
            host: redisConfig.host || 'localhost',
            port: redisConfig.port || 6379,
            password: redisConfig.password,
            db: redisConfig.db || 0,
            ...redisConfig
        });

        this.subscriber = new Redis({
            host: redisConfig.host || 'localhost',
            port: redisConfig.port || 6379,
            password: redisConfig.password,
            db: redisConfig.db || 0,
            ...redisConfig
        });

        this.subscriptions = new Map();
        this.patternSubscriptions = new Map();
        
        this.setupSubscriber();
    }

    setupSubscriber() {
        this.subscriber.on('message', (channel, message) => {
            try {
                const data = JSON.parse(message);
                const handlers = this.subscriptions.get(channel) || [];
                
                handlers.forEach(handler => {
                    try {
                        handler(data.payload, data);
                    } catch (error) {
                        console.error(`Handler error for channel ${channel}:`, error);
                        this.emit('error', error, { channel, message: data });
                    }
                });
            } catch (error) {
                console.error(`Failed to parse message from channel ${channel}:`, error);
            }
        });

        this.subscriber.on('pmessage', (pattern, channel, message) => {
            try {
                const data = JSON.parse(message);
                const handlers = this.patternSubscriptions.get(pattern) || [];
                
                handlers.forEach(handler => {
                    try {
                        handler(data.payload, data, { pattern, channel });
                    } catch (error) {
                        console.error(`Pattern handler error for ${pattern}:`, error);
                        this.emit('error', error, { pattern, channel, message: data });
                    }
                });
            } catch (error) {
                console.error(`Failed to parse pattern message from ${pattern}:`, error);
            }
        });
    }

    async publish(channel, payload, options = {}) {
        const message = {
            id: this.generateMessageId(),
            payload,
            timestamp: Date.now(),
            publisher: options.publisher || 'anonymous',
            type: options.type || 'message'
        };

        try {
            const subscribers = await this.publisher.publish(channel, JSON.stringify(message));
            console.log(`Published message ${message.id} to channel ${channel}, ${subscribers} subscribers`);
            return { messageId: message.id, subscribers };
        } catch (error) {
            console.error(`Failed to publish to channel ${channel}:`, error);
            throw error;
        }
    }

    async subscribe(channel, handler) {
        if (!this.subscriptions.has(channel)) {
            this.subscriptions.set(channel, []);
            await this.subscriber.subscribe(channel);
            console.log(`Subscribed to channel ${channel}`);
        }

        this.subscriptions.get(channel).push(handler);
        return () => this.unsubscribe(channel, handler);
    }

    async unsubscribe(channel, handler = null) {
        if (!this.subscriptions.has(channel)) {
            return;
        }

        if (handler) {
            const handlers = this.subscriptions.get(channel);
            const index = handlers.indexOf(handler);
            if (index > -1) {
                handlers.splice(index, 1);
            }

            if (handlers.length === 0) {
                this.subscriptions.delete(channel);
                await this.subscriber.unsubscribe(channel);
                console.log(`Unsubscribed from channel ${channel}`);
            }
        } else {
            this.subscriptions.delete(channel);
            await this.subscriber.unsubscribe(channel);
            console.log(`Unsubscribed from channel ${channel}`);
        }
    }

    async psubscribe(pattern, handler) {
        if (!this.patternSubscriptions.has(pattern)) {
            this.patternSubscriptions.set(pattern, []);
            await this.subscriber.psubscribe(pattern);
            console.log(`Subscribed to pattern ${pattern}`);
        }

        this.patternSubscriptions.get(pattern).push(handler);
        return () => this.punsubscribe(pattern, handler);
    }

    async punsubscribe(pattern, handler = null) {
        if (!this.patternSubscriptions.has(pattern)) {
            return;
        }

        if (handler) {
            const handlers = this.patternSubscriptions.get(pattern);
            const index = handlers.indexOf(handler);
            if (index > -1) {
                handlers.splice(index, 1);
            }

            if (handlers.length === 0) {
                this.patternSubscriptions.delete(pattern);
                await this.subscriber.punsubscribe(pattern);
                console.log(`Unsubscribed from pattern ${pattern}`);
            }
        } else {
            this.patternSubscriptions.delete(pattern);
            await this.subscriber.punsubscribe(pattern);
            console.log(`Unsubscribed from pattern ${pattern}`);
        }
    }

    async getChannelInfo() {
        const channels = await this.publisher.pubsub('channels');
        const info = {};

        for (const channel of channels) {
            const subscribers = await this.publisher.pubsub('numsub', channel);
            info[channel] = subscribers[1]; // subscribers[0] is channel name, subscribers[1] is count
        }

        return info;
    }

    async disconnect() {
        await Promise.all([
            this.publisher.disconnect(),
            this.subscriber.disconnect()
        ]);
        console.log('Redis PubSub disconnected');
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = RedisPubSub;

Redis Stream

// redis-stream.js - Redis Stream实现
const Redis = require('ioredis');

class RedisStream {
    constructor(streamName, redisConfig = {}) {
        this.streamName = streamName;
        this.redis = new Redis({
            host: redisConfig.host || 'localhost',
            port: redisConfig.port || 6379,
            password: redisConfig.password,
            db: redisConfig.db || 0,
            ...redisConfig
        });

        this.consumerGroup = null;
        this.consumerName = null;
        this.isConsuming = false;
    }

    async add(data, options = {}) {
        const fields = [];
        
        // 添加元数据
        fields.push('id', this.generateMessageId());
        fields.push('timestamp', Date.now());
        fields.push('data', JSON.stringify(data));
        
        if (options.type) {
            fields.push('type', options.type);
        }
        
        if (options.source) {
            fields.push('source', options.source);
        }

        try {
            const messageId = await this.redis.xadd(
                this.streamName,
                options.maxLength ? 'MAXLEN' : '',
                options.maxLength ? '~' : '',
                options.maxLength || '',
                options.id || '*',
                ...fields
            );

            console.log(`Added message ${messageId} to stream ${this.streamName}`);
            return messageId;
        } catch (error) {
            console.error(`Failed to add message to stream ${this.streamName}:`, error);
            throw error;
        }
    }

    async createConsumerGroup(groupName, startId = '$') {
        this.consumerGroup = groupName;
        
        try {
            await this.redis.xgroup('CREATE', this.streamName, groupName, startId, 'MKSTREAM');
            console.log(`Consumer group ${groupName} created for stream ${this.streamName}`);
        } catch (error) {
            if (error.message.includes('BUSYGROUP')) {
                console.log(`Consumer group ${groupName} already exists`);
            } else {
                console.error(`Failed to create consumer group ${groupName}:`, error);
                throw error;
            }
        }
    }

    async consume(consumerName, handler, options = {}) {
        if (!this.consumerGroup) {
            throw new Error('Consumer group not set. Call createConsumerGroup first.');
        }

        this.consumerName = consumerName;
        this.isConsuming = true;

        const count = options.count || 1;
        const block = options.block || 1000;
        const noAck = options.noAck || false;

        console.log(`Consumer ${consumerName} started consuming from stream ${this.streamName}`);

        while (this.isConsuming) {
            try {
                // 首先处理待确认的消息
                await this.processPendingMessages(handler, noAck);

                // 读取新消息
                const result = await this.redis.xreadgroup(
                    'GROUP', this.consumerGroup, consumerName,
                    'COUNT', count,
                    'BLOCK', block,
                    'STREAMS', this.streamName, '>'
                );

                if (result && result.length > 0) {
                    const [streamName, messages] = result[0];
                    
                    for (const [messageId, fields] of messages) {
                        await this.processMessage(messageId, fields, handler, noAck);
                    }
                }
            } catch (error) {
                console.error(`Consumer ${consumerName} error:`, error);
                // 短暂休息后继续
                await new Promise(resolve => setTimeout(resolve, 1000));
            }
        }
    }

    async processMessage(messageId, fields, handler, noAck) {
        const startTime = Date.now();
        
        try {
            // 解析消息字段
            const message = this.parseFields(fields);
            message.id = messageId;
            message.stream = this.streamName;

            console.log(`Processing message ${messageId} from stream ${this.streamName}`);

            // 执行处理器
            await handler(message.data, message);

            // 确认消息
            if (!noAck) {
                await this.redis.xack(this.streamName, this.consumerGroup, messageId);
            }

            const processingTime = Date.now() - startTime;
            console.log(`Message ${messageId} processed in ${processingTime}ms`);

        } catch (error) {
            console.error(`Failed to process message ${messageId}:`, error);
            
            // 可以实现重试逻辑或将消息发送到死信队列
            this.handleProcessingError(messageId, fields, error);
        }
    }

    async processPendingMessages(handler, noAck) {
        try {
            // 获取待确认的消息
            const pending = await this.redis.xpending(
                this.streamName, 
                this.consumerGroup, 
                '-', '+', 10, 
                this.consumerName
            );

            if (pending && pending.length > 0) {
                const messageIds = pending.map(p => p[0]);
                
                const messages = await this.redis.xclaim(
                    this.streamName,
                    this.consumerGroup,
                    this.consumerName,
                    60000, // 1分钟超时
                    ...messageIds
                );

                for (const [messageId, fields] of messages) {
                    await this.processMessage(messageId, fields, handler, noAck);
                }
            }
        } catch (error) {
            console.error('Failed to process pending messages:', error);
        }
    }

    parseFields(fields) {
        const message = {};
        
        for (let i = 0; i < fields.length; i += 2) {
            const key = fields[i];
            const value = fields[i + 1];
            
            if (key === 'data') {
                try {
                    message[key] = JSON.parse(value);
                } catch (error) {
                    message[key] = value;
                }
            } else if (key === 'timestamp') {
                message[key] = parseInt(value);
            } else {
                message[key] = value;
            }
        }
        
        return message;
    }

    handleProcessingError(messageId, fields, error) {
        // 可以实现以下错误处理策略：
        // 1. 重试机制
        // 2. 死信队列
        // 3. 错误日志记录
        console.error(`Message ${messageId} processing error:`, error);
    }

    async read(startId = '0', count = 10) {
        try {
            const result = await this.redis.xread('COUNT', count, 'STREAMS', this.streamName, startId);
            
            if (result && result.length > 0) {
                const [streamName, messages] = result[0];
                return messages.map(([messageId, fields]) => ({
                    id: messageId,
                    ...this.parseFields(fields)
                }));
            }
            
            return [];
        } catch (error) {
            console.error(`Failed to read from stream ${this.streamName}:`, error);
            throw error;
        }
    }

    async getInfo() {
        try {
            const info = await this.redis.xinfo('STREAM', this.streamName);
            const groups = await this.redis.xinfo('GROUPS', this.streamName);
            
            return {
                length: info[1],
                radixTreeKeys: info[3],
                radixTreeNodes: info[5],
                groups: groups.map(group => ({
                    name: group[1],
                    consumers: group[3],
                    pending: group[5],
                    lastDeliveredId: group[7]
                }))
            };
        } catch (error) {
            console.error(`Failed to get stream info for ${this.streamName}:`, error);
            throw error;
        }
    }

    async trim(maxLength, approximate = true) {
        try {
            const trimmed = await this.redis.xtrim(
                this.streamName,
                'MAXLEN',
                approximate ? '~' : '',
                maxLength
            );
            
            console.log(`Trimmed ${trimmed} messages from stream ${this.streamName}`);
            return trimmed;
        } catch (error) {
            console.error(`Failed to trim stream ${this.streamName}:`, error);
            throw error;
        }
    }

    async stop() {
        this.isConsuming = false;
        console.log(`Consumer ${this.consumerName} stopped`);
    }

    async disconnect() {
        this.isConsuming = false;
        await this.redis.disconnect();
        console.log('Redis Stream disconnected');
    }

    generateMessageId() {
        return `msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
    }
}

module.exports = RedisStream;

5. 消息队列选型指南

技术对比

特性	RabbitMQ	Apache Kafka	Redis
消息模型	队列、交换机	主题、分区	列表、发布订阅、流
持久化	支持	支持	支持
性能	中等	高	高
可靠性	高	高	中等
复杂度	中等	高	低
生态系统	丰富	丰富	中等
适用场景	企业级应用	大数据、流处理	缓存、简单队列

选型建议

选择RabbitMQ的场景

• 需要复杂的路由规则
• 要求高可靠性和事务支持
• 企业级应用集成
• 消息量适中（< 100K/秒）

选择Kafka的场景

• 高吞吐量需求（> 100K/秒）
• 流数据处理
• 日志收集和分析
• 事件溯源架构

选择Redis的场景

• 简单的队列需求
• 低延迟要求
• 已有Redis基础设施
• 轻量级应用

6. 性能优化

生产者优化

// producer-optimization.js - 生产者性能优化
class OptimizedProducer {
    constructor(client, options = {}) {
        this.client = client;
        this.batchSize = options.batchSize || 100;
        this.batchTimeout = options.batchTimeout || 100;
        this.compressionType = options.compressionType || 'gzip';
        
        this.messageBuffer = [];
        this.batchTimer = null;
        this.metrics = {
            messagesSent: 0,
            batchesSent: 0,
            errors: 0,
            avgBatchSize: 0,
            avgLatency: 0
        };
    }

    async send(topic, message, options = {}) {
        const messageWithMetadata = {
            topic,
            message,
            timestamp: Date.now(),
            ...options
        };

        this.messageBuffer.push(messageWithMetadata);

        // 检查是否需要立即发送
        if (this.messageBuffer.length >= this.batchSize) {
            await this.flushBatch();
        } else if (!this.batchTimer) {
            // 设置定时器，确保消息不会无限期等待
            this.batchTimer = setTimeout(() => {
                this.flushBatch();
            }, this.batchTimeout);
        }
    }

    async flushBatch() {
        if (this.messageBuffer.length === 0) return;

        const batch = this.messageBuffer.splice(0);
        const startTime = Date.now();

        if (this.batchTimer) {
            clearTimeout(this.batchTimer);
            this.batchTimer = null;
        }

        try {
            // 按主题分组消息
            const messagesByTopic = batch.reduce((acc, msg) => {
                if (!acc[msg.topic]) acc[msg.topic] = [];
                acc[msg.topic].push(msg.message);
                return acc;
            }, {});

            // 并行发送到不同主题
            const sendPromises = Object.entries(messagesByTopic).map(
                ([topic, messages]) => this.client.sendBatch(topic, messages, {
                    compression: this.compressionType
                })
            );

            await Promise.all(sendPromises);

            // 更新指标
            const latency = Date.now() - startTime;
            this.updateMetrics(batch.length, latency);

            console.log(`Sent batch of ${batch.length} messages in ${latency}ms`);

        } catch (error) {
            console.error('Batch send failed:', error);
            this.metrics.errors++;
            
            // 重试逻辑
            await this.retryBatch(batch);
        }
    }

    async retryBatch(batch, attempt = 1) {
        const maxRetries = 3;
        const retryDelay = Math.pow(2, attempt) * 1000;

        if (attempt > maxRetries) {
            console.error(`Batch failed after ${maxRetries} retries`);
            return;
        }

        console.log(`Retrying batch in ${retryDelay}ms (attempt ${attempt})`);
        
        setTimeout(async () => {
            try {
                // 重新发送批次
                this.messageBuffer.unshift(...batch);
                await this.flushBatch();
            } catch (error) {
                await this.retryBatch(batch, attempt + 1);
            }
        }, retryDelay);
    }

    updateMetrics(batchSize, latency) {
        this.metrics.messagesSent += batchSize;
        this.metrics.batchesSent++;
        this.metrics.avgBatchSize = this.metrics.messagesSent / this.metrics.batchesSent;
        this.metrics.avgLatency = (this.metrics.avgLatency + latency) / 2;
    }

    getMetrics() {
        return { ...this.metrics };
    }

    async close() {
        // 发送剩余消息
        await this.flushBatch();
        
        if (this.batchTimer) {
            clearTimeout(this.batchTimer);
        }
    }
}

module.exports = OptimizedProducer;

消费者优化

// consumer-optimization.js - 消费者性能优化
class OptimizedConsumer {
    constructor(client, options = {}) {
        this.client = client;
        this.concurrency = options.concurrency || 10;
        this.batchSize = options.batchSize || 50;
        this.prefetchCount = options.prefetchCount || this.concurrency * 2;
        
        this.workers = [];
        this.messageQueue = [];
        this.isProcessing = false;
        
        this.metrics = {
            messagesProcessed: 0,
            messagesPerSecond: 0,
            avgProcessingTime: 0,
            errors: 0,
            lastResetTime: Date.now()
        };

        // 定期重置速率指标
        setInterval(() => this.resetRateMetrics(), 60000);
    }

    async startConsumer(topics, processor, options = {}) {
        this.isProcessing = true;

        // 设置预取数量
        await this.client.setPrefetch(this.prefetchCount);

        // 启动工作线程池
        for (let i = 0; i < this.concurrency; i++) {
            this.workers.push(this.startWorker(processor, i));
        }

        // 启动消息获取循环
        this.startMessageFetcher(topics, options);

        console.log(`Started optimized consumer with ${this.concurrency} workers`);
    }

    async startMessageFetcher(topics, options) {
        while (this.isProcessing) {
            try {
                // 批量获取消息
                const messages = await this.client.fetchMessages(topics, {
                    maxMessages: this.batchSize,
                    timeout: options.fetchTimeout || 1000
                });

                if (messages && messages.length > 0) {
                    // 将消息添加到内部队列
                    this.messageQueue.push(...messages);
                    console.log(`Fetched ${messages.length} messages`);
                }

                // 如果队列过长，暂停获取
                if (this.messageQueue.length > this.prefetchCount * 2) {
                    await new Promise(resolve => setTimeout(resolve, 100));
                }

            } catch (error) {
                console.error('Message fetcher error:', error);
                await new Promise(resolve => setTimeout(resolve, 1000));
            }
        }
    }

    async startWorker(processor, workerId) {
        while (this.isProcessing) {
            try {
                // 从内部队列获取消息
                const message = this.messageQueue.shift();
                
                if (!message) {
                    // 队列为空，短暂等待
                    await new Promise(resolve => setTimeout(resolve, 10));
                    continue;
                }

                const startTime = Date.now();

                try {
                    // 处理消息
                    await processor(message.data, message);
                    
                    // 确认消息
                    await this.client.ackMessage(message);
                    
                    const processingTime = Date.now() - startTime;
                    this.updateMetrics(processingTime);
                    
                    console.log(`Worker ${workerId} processed message in ${processingTime}ms`);

                } catch (error) {
                    console.error(`Worker ${workerId} processing error:`, error);
                    this.metrics.errors++;
                    
                    // 错误处理
                    await this.handleProcessingError(message, error);
                }

            } catch (error) {
                console.error(`Worker ${workerId} error:`, error);
                await new Promise(resolve => setTimeout(resolve, 1000));
            }
        }
    }

    async handleProcessingError(message, error) {
        // 实现重试逻辑
        if (!message.retryCount) message.retryCount = 0;
        
        if (message.retryCount < 3) {
            message.retryCount++;
            const delay = Math.pow(2, message.retryCount) * 1000;
            
            setTimeout(() => {
                this.messageQueue.unshift(message);
            }, delay);
            
            console.log(`Message will retry in ${delay}ms (attempt ${message.retryCount})`);
        } else {
            // 发送到死信队列
            await this.client.sendToDeadLetter(message);
            console.log('Message sent to dead letter queue');
        }
    }

    updateMetrics(processingTime) {
        this.metrics.messagesProcessed++;
        this.metrics.avgProcessingTime = 
            (this.metrics.avgProcessingTime + processingTime) / 2;
    }

    resetRateMetrics() {
        const now = Date.now();
        const timeDiff = (now - this.metrics.lastResetTime) / 1000;
        
        this.metrics.messagesPerSecond = this.metrics.messagesProcessed / timeDiff;
        this.metrics.messagesProcessed = 0;
        this.metrics.lastResetTime = now;
    }

    getMetrics() {
        return {
            ...this.metrics,
            queueLength: this.messageQueue.length,
            activeWorkers: this.workers.length
        };
    }

    async stop() {
        this.isProcessing = false;
        
        // 等待所有工作线程完成
        await Promise.all(this.workers);
        
        console.log('Optimized consumer stopped');
    }
}

module.exports = OptimizedConsumer;

7. 监控和告警

监控指标收集

// monitoring.js - 消息队列监控
const prometheus = require('prom-client');

class MessageQueueMonitoring {
    constructor() {
        // 创建指标
        this.messagesSentTotal = new prometheus.Counter({
            name: 'mq_messages_sent_total',
            help: 'Total number of messages sent',
            labelNames: ['queue', 'topic', 'producer']
        });

        this.messagesReceivedTotal = new prometheus.Counter({
            name: 'mq_messages_received_total',
            help: 'Total number of messages received',
            labelNames: ['queue', 'topic', 'consumer']
        });

        this.messageProcessingDuration = new prometheus.Histogram({
            name: 'mq_message_processing_duration_seconds',
            help: 'Message processing duration in seconds',
            labelNames: ['queue', 'topic', 'consumer'],
            buckets: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10]
        });

        this.queueSize = new prometheus.Gauge({
            name: 'mq_queue_size',
            help: 'Current queue size',
            labelNames: ['queue', 'type']
        });

        this.consumerLag = new prometheus.Gauge({
            name: 'mq_consumer_lag',
            help: 'Consumer lag in messages',
            labelNames: ['queue', 'topic', 'consumer_group']
        });

        this.errorRate = new prometheus.Counter({
            name: 'mq_errors_total',
            help: 'Total number of errors',
            labelNames: ['queue', 'topic', 'error_type']
        });

        this.connectionStatus = new prometheus.Gauge({
            name: 'mq_connection_status',
            help: 'Connection status (1 = connected, 0 = disconnected)',
            labelNames: ['queue', 'type']
        });

        // 启动指标收集
        this.startMetricsCollection();
    }

    recordMessageSent(queue, topic, producer = 'unknown') {
        this.messagesSentTotal.inc({ queue, topic, producer });
    }

    recordMessageReceived(queue, topic, consumer = 'unknown') {
        this.messagesReceivedTotal.inc({ queue, topic, consumer });
    }

    recordProcessingTime(queue, topic, consumer, duration) {
        this.messageProcessingDuration
            .labels({ queue, topic, consumer })
            .observe(duration / 1000); // 转换为秒
    }

    updateQueueSize(queue, type, size) {
        this.queueSize.set({ queue, type }, size);
    }

    updateConsumerLag(queue, topic, consumerGroup, lag) {
        this.consumerLag.set({ queue, topic, consumer_group: consumerGroup }, lag);
    }

    recordError(queue, topic, errorType) {
        this.errorRate.inc({ queue, topic, error_type: errorType });
    }

    updateConnectionStatus(queue, type, status) {
        this.connectionStatus.set({ queue, type }, status ? 1 : 0);
    }

    startMetricsCollection() {
        // 定期收集系统指标
        setInterval(async () => {
            try {
                await this.collectSystemMetrics();
            } catch (error) {
                console.error('Failed to collect system metrics:', error);
            }
        }, 30000); // 每30秒收集一次
    }

    async collectSystemMetrics() {
        // 这里可以集成各种消息队列的管理API
        // 收集队列大小、消费者状态等信息
        
        // 示例：RabbitMQ管理API
        // const rabbitMQStats = await this.getRabbitMQStats();
        // this.updateQueueSize('rabbitmq', 'waiting', rabbitMQStats.waiting);
        
        // 示例：Kafka管理API  
        // const kafkaStats = await this.getKafkaStats();
        // this.updateConsumerLag('kafka', 'user-events', 'user-service', kafkaStats.lag);
    }

    getMetrics() {
        return prometheus.register.metrics();
    }

    createAlertRules() {
        return {
            // 队列积压告警
            queueBacklog: {
                expr: 'mq_queue_size{type="waiting"} > 1000',
                for: '5m',
                labels: { severity: 'warning' },
                annotations: {
                    summary: 'Message queue backlog detected',
                    description: 'Queue {{ $labels.queue }} has {{ $value }} pending messages'
                }
            },

            // 消费者延迟告警
            consumerLag: {
                expr: 'mq_consumer_lag > 100',
                for: '2m',
                labels: { severity: 'warning' },
                annotations: {
                    summary: 'Consumer lag detected',
                    description: 'Consumer {{ $labels.consumer_group }} is lagging by {{ $value }} messages'
                }
            },

            // 错误率告警
            errorRate: {
                expr: 'rate(mq_errors_total[5m]) > 0.1',
                for: '1m',
                labels: { severity: 'critical' },
                annotations: {
                    summary: 'High error rate detected',
                    description: 'Error rate is {{ $value }} errors per second'
                }
            },

            // 连接状态告警
            connectionDown: {
                expr: 'mq_connection_status == 0',
                for: '30s',
                labels: { severity: 'critical' },
                annotations: {
                    summary: 'Message queue connection down',
                    description: 'Connection to {{ $labels.queue }} is down'
                }
            },

            // 处理时间告警
            slowProcessing: {
                expr: 'histogram_quantile(0.95, mq_message_processing_duration_seconds) > 5',
                for: '5m',
                labels: { severity: 'warning' },
                annotations: {
                    summary: 'Slow message processing detected',
                    description: '95th percentile processing time is {{ $value }}s'
                }
            }
        };
    }
}

module.exports = MessageQueueMonitoring;

健康检查

// health-check.js - 消息队列健康检查
class MessageQueueHealthCheck {
    constructor(clients = {}) {
        this.clients = clients;
        this.healthStatus = {};
        this.checkInterval = 30000; // 30秒
        
        this.startHealthChecks();
    }

    startHealthChecks() {
        setInterval(async () => {
            await this.performHealthChecks();
        }, this.checkInterval);

        // 立即执行一次
        this.performHealthChecks();
    }

    async performHealthChecks() {
        const checks = [];

        // RabbitMQ健康检查
        if (this.clients.rabbitmq) {
            checks.push(this.checkRabbitMQ());
        }

        // Kafka健康检查
        if (this.clients.kafka) {
            checks.push(this.checkKafka());
        }

        // Redis健康检查
        if (this.clients.redis) {
            checks.push(this.checkRedis());
        }

        await Promise.all(checks);
        
        console.log('Health check completed:', this.healthStatus);
    }

    async checkRabbitMQ() {
        const startTime = Date.now();
        
        try {
            // 检查连接状态
            if (!this.clients.rabbitmq.isConnected) {
                throw new Error('RabbitMQ not connected');
            }

            // 发送测试消息
            const testQueue = 'health-check-queue';
            await this.clients.rabbitmq.assertQueue(testQueue, { 
                durable: false, 
                autoDelete: true 
            });
            
            const testMessage = { 
                test: true, 
                timestamp: Date.now() 
            };
            
            await this.clients.rabbitmq.sendToQueue(testQueue, testMessage);
            
            // 尝试消费测试消息
            const message = await this.clients.rabbitmq.get(testQueue);
            if (!message) {
                throw new Error('Failed to retrieve test message');
            }

            const responseTime = Date.now() - startTime;
            
            this.healthStatus.rabbitmq = {
                status: 'healthy',
                responseTime,
                timestamp: new Date().toISOString(),
                details: {
                    connected: true,
                    messageRoundTrip: true
                }
            };

        } catch (error) {
            this.healthStatus.rabbitmq = {
                status: 'unhealthy',
                error: error.message,
                timestamp: new Date().toISOString(),
                details: {
                    connected: this.clients.rabbitmq.isConnected || false
                }
            };
        }
    }

    async checkKafka() {
        const startTime = Date.now();
        
        try {
            // 检查生产者和消费者连接
            const producer = this.clients.kafka.producer;
            const admin = this.clients.kafka.admin;

            // 获取集群元数据
            const metadata = await admin.fetchTopicMetadata({
                topics: ['__consumer_offsets'] // 系统主题，通常存在
            });

            // 检查broker连接
            if (!metadata.brokers || metadata.brokers.length === 0) {
                throw new Error('No brokers available');
            }

            const responseTime = Date.now() - startTime;
            
            this.healthStatus.kafka = {
                status: 'healthy',
                responseTime,
                timestamp: new Date().toISOString(),
                details: {
                    brokers: metadata.brokers.length,
                    connected: true
                }
            };

        } catch (error) {
            this.healthStatus.kafka = {
                status: 'unhealthy',
                error: error.message,
                timestamp: new Date().toISOString(),
                details: {
                    connected: false
                }
            };
        }
    }

    async checkRedis() {
        const startTime = Date.now();
        
        try {
            // 检查Redis连接
            const redis = this.clients.redis;
            
            // 执行PING命令
            const pong = await redis.ping();
            if (pong !== 'PONG') {
                throw new Error('Redis ping failed');
            }

            // 测试读写操作
            const testKey = 'health-check:' + Date.now();
            const testValue = 'test-value';
            
            await redis.set(testKey, testValue, 'EX', 60); // 60秒过期
            const retrievedValue = await redis.get(testKey);
            
            if (retrievedValue !== testValue) {
                throw new Error('Redis read/write test failed');
            }

            await redis.del(testKey);

            const responseTime = Date.now() - startTime;
            
            this.healthStatus.redis = {
                status: 'healthy',
                responseTime,
                timestamp: new Date().toISOString(),
                details: {
                    connected: true,
                    readWrite: true
                }
            };

        } catch (error) {
            this.healthStatus.redis = {
                status: 'unhealthy',
                error: error.message,
                timestamp: new Date().toISOString(),
                details: {
                    connected: false
                }
            };
        }
    }

    getHealthStatus() {
        const overall = Object.values(this.healthStatus).every(
            status => status.status === 'healthy'
        );

        return {
            status: overall ? 'healthy' : 'unhealthy',
            timestamp: new Date().toISOString(),
            services: this.healthStatus
        };
    }

    // Express.js健康检查端点
    createHealthEndpoint() {
        return (req, res) => {
            const health = this.getHealthStatus();
            const statusCode = health.status === 'healthy' ? 200 : 503;
            
            res.status(statusCode).json(health);
        };
    }
}

module.exports = MessageQueueHealthCheck;

8. 最佳实践

消息设计原则

1. 幂等性设计

   • 消息包含唯一ID
   • 消费者实现幂等处理
   • 使用去重机制

2. 消息版本控制

   • 向后兼容的消息格式
   • 版本号标识
   • 渐进式升级策略

3. 错误处理

   • 重试机制
   • 死信队列
   • 错误分类和处理

4. 性能优化

   • 批量处理
   • 连接池管理
   • 合理的预取设置

安全考虑

1. 认证和授权

   • 使用强密码和证书
   • 细粒度权限控制
   • 定期轮换凭据

2. 数据加密

   • 传输加密（TLS）
   • 存储加密
   • 敏感数据脱敏

3. 网络安全

   • 防火墙配置
   • VPN或专网
   • 访问控制列表

运维建议

1. 容量规划

   • 监控资源使用
   • 预测增长趋势
   • 弹性扩缩容

2. 备份和恢复

   • 定期备份配置
   • 数据备份策略
   • 灾难恢复计划

3. 版本管理

   • 滚动升级
   • 蓝绿部署
   • 回滚策略

总结

消息队列是现代分布式系统的重要组件，选择合适的消息队列技术并正确实施，对系统的可靠性、性能和可维护性至关重要。本指南涵盖了主流消息队列技术的实战应用，包括：

1. 技术选型：根据业务需求选择合适的消息队列
2. 实现模式：掌握各种消息传递模式的实现
3. 性能优化：通过批处理、连接池等技术提升性能
4. 监控运维：建立完善的监控和告警机制
5. 最佳实践：遵循安全、可靠、高效的设计原则

通过合理的架构设计和实施，消息队列能够有效解决系统解耦、异步处理、削峰填谷等问题，为构建高可用、高性能的分布式系统提供坚实基础。