基于NodeJS和Redis的分布式异步消息处理服务源码

# NodeResque - 轻量级异步消息处理服务 基于Node.js和Redis的分布式异步消息处理服务，支持多进程多服务器部署，具有消息类型配置、监控告警等功能。 ## 系统架构 ```mermaid graph TD Client[客户端] --> API[RESTful API] API --> Queue[Redis消息队列] Queue --> Processor[消息处理器] Processor --> Email[邮件服务] Processor --> Other[其他服务] Monitor[监控系统] --> Queue Monitor --> Processor Monitor --> Alert[告警系统] ``` ## 主要特性 - 基于Redis的分布式消息队列 - 支持消息优先级和重试机制 - 可配置的消息处理器 - 实时监控和告警系统 - RESTful API接口 - 多进程多服务器部署支持 ## 安装 ```bash npm install ``` ## 配置 在项目根目录创建 `.env` 文件，配置以下环境变量： ```env # Redis配置 REDIS_HOST=localhost REDIS_PORT=6379 REDIS_PASSWORD= # 服务配置 PORT=3000 NODE_ENV=development # 消息处理配置 MAX_RETRY_ATTEMPTS=3 MESSAGE_TIMEOUT=300000 MAX_CONCURRENT_JOBS=5 # 监控配置 ALERT_QUEUE_THRESHOLD=1000 ALERT_PROCESSING_DELAY=60000 ``` ## 开发环境搭建 1. 安装依赖 ```bash npm install ``` 2. 启动Redis服务 ```bash # Windows redis-server # Linux/Mac redis-server /usr/local/etc/redis.conf ``` 3. 启动开发服务器 ```bash npm run dev ``` ## 使用示例 ### 启动服务 ```bash npm start ``` ### 注册消息处理器 ```javascript // 邮件处理器配置示例 // POST /processor/email { "type": "email", "options": { "concurrency": 5, "timeout": 30000, "maxRetries": 3, "smtp": { "host": "smtp.example.com", "port": 587, "secure": true, "auth": { "user": "[email protected]", "pass": "your-password" } } }, "handler": "async function(message) { const { to, subject, content } = message; // 邮件发送逻辑 await sendEmail({ to, subject, content }); return true; }" } ``` ### 发送消息 ```javascript // POST /message/email { "message": { "to": "[email protected]", "subject": "Test Email", "content": "Hello World" }, "priority": 1 } ``` ### 查看队列状态 ```bash GET /stats/email ``` ### 查看监控指标 ```bash GET /metrics ``` ## API接口 - `POST /message/:type` - 发送消息到指定类型的队列 - `POST /processor/:type` - 注册消息处理器 - `GET /stats/:type` - 获取指定队列的统计信息 - `GET /metrics` - 获取监控指标 - `GET /processors` - 获取所有注册的处理器 ## 监控告警 系统会自动监控以下指标： ### 队列监控 - 队列堆积数量（阈值：1000条） - 消息处理延迟（阈值：60秒） - 错误率统计（阈值：5%） ### 处理器监控 - CPU使用率（阈值：80%） - 内存使用率（阈值：85%） - 处理超时率（阈值：10%） ### 告警方式 - 邮件通知 - Webhook回调 - 日志记录 ## 性能调优 ### Redis优化 - 启用持久化（AOF/RDB） - 合理设置内存上限 - 使用Redis集群提高可用性 ### 处理器优化 - 根据消息类型调整并发数 - 设置合理的超时时间 - 实现优雅的错误处理 ### 系统优化 - 使用PM2进行进程管理 - 启用Node.js垃圾回收优化 - 配置合适的日志级别 ## 最佳实践 - 根据业务场景合理设置消息优先级 - 实现幂等的消息处理逻辑 - 定期清理过期的消息数据 - 配置合适的监控告警阈值 - 做好系统容量规划 ## 注意事项 - 确保Redis服务器已启动并可访问 - 建议在生产环境中配置适当的监控阈值 - 根据实际需求调整消息处理器的并发数和超时设置 ## 许可证 MIT