消息队列RabbitMQ代码重构：从烂代码到优雅代码

2016年我接手了一个使用RabbitMQ的项目。

说起来这个项目是公司的一个老项目已经运行了两年了用RabbitMQ做消息队列处理异步任务比如发送邮件发送短信生成报表数据同步，等。

但是这个项目的代码写得很乱问题很多经常出bug维护起来很痛苦。比如有时候消息丢了不知道为什么；有时候消费者挂了没人知道；有时候消息重复消费导致数据错误；等等。

而且代码耦合很严重生产者和消费者的代码混在一起到处都是重复的代码修改一个功能要改很多地方很容易出错。

我接手之后看了几天代码实在看不下去了于是决定花一周时间对RabbitMQ相关的代码进行重构从烂代码变成优雅代码提升可维护性和稳定性。

今天就来聊聊这次代码重构的经历以及消息队列代码设计的一些经验。

一、原来的代码有什么问题

先说说原来的代码有什么问题。

我看了几天代码总结了以下几个主要的问题：

1. 连接管理混乱

原来的代码每次发消息或者消费消息都新建一个RabbitMQ连接用完就关掉没有连接池也没有长连接复用。

这样导致几个问题：

• 性能差：每次都要建立连接握手认证等开销很大性能，差。
• 资源浪费：频繁建立关闭连接浪费系统资源也给RabbitMQ服务器造成压力。
• 不稳定：网络波动的时候连接容易失败而且没有重连机制连接断了就再也连不上了需要手动重启。

2. 生产者代码重复

原来的代码每个需要发消息的地方都写了一遍发消息的逻辑包括建立连接创建通道声明交换机声明队列绑定发送消息关闭连接等到处都是重复的代码。

而且每个地方的代码还不太一样有的声明了交换机有的没声明；有的用了持久化有的没用；有的设置了消息属性有的没设置很不统一很容易出错。

修改一个发消息的逻辑要改很多地方很容易漏改导致有的地方改了有的地方没改行为不一致。

3. 消费者代码混乱

原来的消费者代码更乱每个消费者都是一个独立的PHP脚本在命令行运行代码结构各不相同有的用了while循环有的用了递归；有的有错误处理有的没有；有的有日志有的没有。

而且消费者的代码和业务逻辑混在一起消费消息的代码和处理业务的代码写在一起耦合很严重修改业务逻辑容易影响消费逻辑反之亦然。

消费者也没有统一的管理启动停止重启都要手动操作很不方便而且消费者挂了也没人知道需要人工监控。

4. 错误处理不完善

原来的代码错误处理很不完善很多地方都没有try-catch出了异常直接报错退出没有重试机制也没有降级方案。

比如发消息失败了直接报错消息就丢了没有重试也没有记录导致数据丢失。

消费消息失败了直接报错消费者退出消息也没有重新入队或者进入死信队列导致消息丢失或者卡住。

而且也没有告警机制出了问题没人知道等用户反馈了才发现已经晚了。

5. 没有日志和监控

原来的代码几乎没有日志也没有监控出了问题不知道什么时候出的什么原因只能靠猜排查问题很困难。

比如消息丢了不知道是生产者没发出去还是RabbitMQ丢了还是消费者没消费还是消费失败了没有日志很难排查。

消费者挂了也不知道什么时候挂的什么原因挂的只能看进程在不在没有历史记录。

6. 配置硬编码

原来的代码很多配置都是硬编码的比如RabbitMQ的地址端口用户名密码交换机名称队列名称等都写死在代码里。

这样导致几个问题：

• 环境切换困难：开发测试生产环境的配置不一样要改代码很麻烦也容易出错。
• 安全性差：用户名密码等敏感信息写在代码里不安全容易泄露。
• 维护困难：修改配置要改代码重新部署很不方便。

二、重构的思路

找到了问题就开始思考重构的思路。

我的重构思路主要有以下几点：

1. 分层设计

把RabbitMQ相关的代码分成几层职责清晰耦合降低：

• 连接层：负责RabbitMQ连接的管理包括连接池长连接重连，等。
• 生产者层：负责消息的发送封装统一的发消息接口。
• 消费者层：负责消息的消费封装统一的消费者基类和管理。
• 业务层：负责具体的业务逻辑和RabbitMQ解耦。

这样每层职责清晰修改一层不会影响其他层可维护性大幅提升。

2. 单一职责

每个类每个方法都只做一件事职责单一不要把多个逻辑混在一起。

比如连接管理的类只负责连接的管理不负责发消息；发消息的类只负责发消息不负责连接管理；消费者基类只负责消费的通用逻辑不负责具体的业务。

这样代码更清晰更易读也更易维护。

3. 复用和封装

把重复的代码封装成公共的类和方法复用避免到处都是重复的代码。

比如发消息的逻辑封装成一个Producer类需要发消息的地方都调用这个类的方法不用每个地方都写一遍。

消费者的通用逻辑封装成一个Consumer基类具体的消费者继承这个基类只实现业务逻辑不用每个消费者都写一遍消费的通用逻辑。

4. 完善错误处理

完善错误处理每个可能出错的地方都有try-catch有重试机制有降级方案有告警机制。

比如发消息失败了自动重试几次还失败就记录日志告警通知开发人员同时把消息存到本地或者数据库等恢复了再补发。

消费消息失败了自动重试几次还失败就把消息放到死信队列或者记录日志告警通知开发人员人工处理不要让消息丢失。

5. 完善日志和监控

完善日志和监控每个关键步骤都有日志出了问题能快速定位原因。

比如发消息的时候记录消息的内容交换机路由键发送结果，等；消费消息的时候记录消息的内容消费开始时间结束时间消费结果，等。

同时增加监控比如监控队列的消息堆积数量消费者的运行状态消息的消费成功率等出了异常自动告警。

6. 配置外置

把配置从代码里抽出来放到配置文件里不要硬编码。

比如RabbitMQ的地址端口用户名密码交换机名称队列名称等都放到配置文件里代码从配置文件读取配置。

这样环境切换修改配置都很方便不用改代码也更安全。

三、重构的具体步骤

思路明确了就开始具体的重构。

1. 封装连接管理类

首先封装一个RabbitMQ连接管理类负责连接的创建复用重连，等。

这个类的主要功能：

• 单例模式：整个应用只有一个连接实例避免重复创建连接。
• 长连接：连接建立之后保持长连接复用不用每次都新建。
• 自动重连：连接断了自动重连不用手动重启。
• 通道管理：管理通道的创建和复用避免频繁创建通道。

用了这个连接管理类之后连接的问题就解决了性能提升了稳定性也提升了。

2. 封装生产者类

然后封装一个生产者类负责消息的发送提供统一的发消息接口。

这个类的主要功能：

• 统一的发消息接口：提供一个send方法传入交换机路由键消息内容等就能发消息不用关心底层细节。
• 消息持久化：默认消息持久化RabbitMQ重启消息也不会，丢。
• 发送确认：使用发布确认机制确保消息成功到达RabbitMQ没有，丢。
• 失败重试：发消息失败了自动重试几次还失败就记录日志告警。
• 消息属性：统一设置消息属性比如内容类型持久化过期时间，等。

用了这个生产者类之后发消息的代码就统一了不再到处都是重复的代码修改发消息的逻辑只要改这一个类就行。

3. 封装消费者基类

然后封装一个消费者基类负责消费的通用逻辑具体的消费者继承这个基类只实现业务逻辑。

这个基类的主要功能：

• 统一的消费流程：封装连接声明队列绑定消费等通用流程子类不用关心。
• 消息确认：消费成功手动确认消息；消费失败不确认消息重新入队或者进入死信队列。
• 错误处理：消费过程中出了异常自动捕获记录日志重试或者进入死信队列不会让消费者退出。
• 自动重连：连接断了自动重连继续消费不用手动重启。
• 日志记录：消费的每个步骤都有日志出了问题能快速定位。
• 优雅退出：收到退出信号优雅退出处理完当前消息再退出不会丢消息。

具体的消费者只需要继承这个基类实现一个handle方法处理具体的业务逻辑就行不用关心消费的通用逻辑。

这样消费者的代码就统一了结构清晰业务逻辑和消费逻辑解耦可维护性大幅提升。

4. 封装消费者管理类

然后封装一个消费者管理类负责消费者的启动停止重启监控，等。

这个类的主要功能：

• 统一启动：一个命令就能启动所有消费者不用一个个手动启动。
• 统一停止：一个命令就能停止所有消费者优雅退出。
• 统一重启：一个命令就能重启所有消费者。
• 状态监控：监控所有消费者的运行状态消费者挂了自动重启同时告警。
• 进程管理：管理消费者的进程记录进程ID方便管理。

用了这个消费者管理类之后消费者的管理就方便了很多不用手动一个个操作也不怕消费者挂了没人知道。

5. 业务逻辑和消息队列解耦

然后把业务逻辑和消息队列解耦业务代码不直接依赖RabbitMQ的细节。

具体做法：

• 定义统一的消息格式比如消息包含类型数据时间戳等业务代码只关心消息的数据不关心RabbitMQ的细节。
• 生产者只负责把消息发到RabbitMQ不关心谁来消费怎么消费。
• 消费者只负责从RabbitMQ取消息调用业务处理类处理消息不关心业务的具体逻辑。
• 业务处理类只负责处理具体的业务逻辑不关心消息来自哪里怎么来的。

这样业务逻辑和消息队列完全解耦以后想换消息队列比如从RabbitMQ换成Kafka或者Redis只要改生产者和消费者的代码业务代码不用改很方便。

6. 完善日志和监控

然后完善日志和监控。

日志方面：

• 生产者发消息记录日志：消息内容交换机路由键发送结果耗时，等。
• 消费者消费消息记录日志：消息内容消费开始时间结束时间耗时消费结果，等。
• 错误记录日志：出了错误记录错误信息堆栈上下文等方便排查。
• 日志分级：DEBUG，INFO，WARNING，ERROR等不同级别的日志方便筛选。

监控方面：

• 队列监控：监控每个队列的消息数量消费者数量等消息堆积过多告警。
• 消费者监控：监控每个消费者的运行状态消费速度成功率等消费者挂了或者消费失败率过高告警。
• 消息监控：监控消息的发送成功率消费成功率延迟等出了异常告警。
• 告警方式：邮件短信微信等多种方式确保开发人员能及时收到告警。

7. 配置外置

最后把配置从代码里抽出来放到配置文件，里。

配置文件包含：

• RabbitMQ连接配置：地址端口用户名密码虚拟主机，等。
• 交换机配置：名称类型是否持久化，等。
• 队列配置：名称是否持久化是否排他是否自动删除，等。
• 绑定配置：队列和交换机的绑定关系路由键，等。
• 消费者配置：消费者类名队列名称并发数，等。
• 重试配置：重试次数重试间隔，等。
• 死信配置：死信交换机死信队列，等。

代码从配置文件读取配置不用硬编码环境切换修改配置都很方便。

四、踩坑经验

在重构的过程中也踩了一些坑分享一下。

1. 长连接断开问题

刚开始用长连接的时候发现连接经常莫名其妙地断开后来发现是因为TCP的keepalive时间太长或者防火墙会自动断开空闲的连接。

解决方法：

• 设置RabbitMQ的心跳间隔比如30秒定期发送心跳保持连接活跃。
• 设置TCP的keepalive参数缩短keepalive时间定期检测连接是否还在。
• 实现自动重连机制连接断了自动重连继续工作。

2. 消息重复消费问题

重构之后发现有时候消息会被重复消费导致数据错误。后来发现是因为消费者消费消息之后还没来得及确认消费者就挂了或者连接断了消息重新入队被其他消费者再次消费。

解决方法：

• 业务逻辑做幂等处理即使消息被重复消费也不会导致数据错误。比如用唯一的消息ID判断是否已经消费过消费过就直接跳过。
• 消费成功之后立即确认消息不要做太多其他操作再确认减少确认之前出问题的概率。
• 对于重要的消息可以用事务或者分布式锁保证消息只被消费一次。

3. 消息堆积问题

重构之后有一次消费者出了bug消费失败消息一直重新入队导致队列消息堆积越来越多最后RabbitMQ内存满了影响了其他业务。

解决方法：

• 设置消息的最大重试次数超过次数就进入死信队列不要一直重新入队。
• 监控队列的消息数量堆积过多自动告警及时处理。
• 设置RabbitMQ的内存告警和磁盘告警快满了提前告警。
• 消费者做好错误处理不要因为一个消息消费失败就卡住影响其他消息。

4. 死信队列配置问题

刚开始配置死信队列的时候发现消息没有进入死信队列还是一直重新入队。后来发现是因为死信队列的配置不对队列没有设置死信交换机和死信路由键。

解决方法：

• 声明队列的时候设置x-dead-letter-exchange和x-dead-letter-routing-key参数指定死信交换机和路由键。
• 声明死信交换机和死信队列绑定确保死信能正确路由到死信队列。
• 测试验证消息过期或者被拒绝之后能正确进入死信队列。

5. 发布确认机制问题

刚开始用发布确认机制的时候发现有时候消息发出去了但是没有收到确认以为丢了实际上没丢是确认延迟了。

解决方法：

• 发布确认是异步的发完消息不要立即等确认可以批量等确认提高性能。
• 设置确认超时时间超时没收到确认再重试不要等太久。
• 记录未确认的消息定期检查超时的消息重试或者告警。

五、重构后的效果

经过一周的重构最终效果如下：

1. 代码质量提升

• 代码结构清晰分层合理职责单一易读易维护。
• 重复代码大幅减少原来到处都是重复的代码现在都封装成公共的类和方法复用。
• 业务逻辑和消息队列解耦修改业务不会影响消息队列修改消息队列不会影响业务。
• 配置外置环境切换修改配置都很方便。

2. 稳定性提升

• 长连接，+，自动重连连接断了自动恢复不用手动重启。
• 完善错误处理出了异常自动重试或者进入死信队列不会丢消息也不会让消费者退出。
• 消息持久化，+，发布确认确保消息不丢。
• 消费者挂了自动重启，+，告警及时发现问题。

3. 性能提升

• 长连接复用不用每次都新建连接性能提升很多。
• 通道复用不用每次都新建通道性能提升。
• 批量发送批量确认提高吞吐量。

4. 可维护性提升

• 统一的代码结构和规范新人上手，快。
• 完善日志出了问题能快速定位原因。
• 完善监控出了异常自动告警及时处理。
• 统一的消费者管理启动停止重启都很方便。

重构之后这个项目的RabbitMQ相关的代码从烂代码变成了优雅代码维护起来轻松了很多出问题的次数也大幅减少老板和同事都很满意。

六、代码重构的经验

通过这次RabbitMQ代码重构我总结了一些代码重构的经验。

1. 重构不是重写

重构不是把原来的代码全部扔掉重新写而是在不改变外部行为的前提下改善代码的内部结构。

重构要小步快跑一步一步来每改一步都要测试确保功能正常不要一下子改太多出了问题不知道是哪里改的。

2. 先理解再重构

重构之前一定要先理解原来的代码知道它是怎么工作的有什么功能有什么坑然后再开始重构。

不要没理解就盲目重构那样很容易改出问题把原来正常的功能改坏了。

3. 有测试再重构

重构之前最好有测试用例确保重构之后功能还是正常的。如果没有测试用例先补测试用例再重构。

有了测试用例重构的时候就有信心改完跑一下测试就知道有没有改坏。

4. 持续重构

重构不是一次性的工作而是持续的过程。平时写代码的时候看到不好的代码就顺手重构一下不要等代码烂到看不下去了才来重构那样成本很高。

所谓，"随时重构随地重构"，就是这个道理。

5. 追求代码的优雅

好的代码不仅能运行还要易读易维护易扩展这就是代码的优雅。

写代码的时候要追求代码的优雅不要满足于能跑就行要多想想有没有更好的写法更清晰的结构更合理的设计。

代码是写给人看的顺便能在机器上运行所以可读性很重要。

七、写在最后

消息队列RabbitMQ代码重构：从烂代码到优雅代码。

这次代码重构虽然只有一周时间但是收获很大不仅把项目的代码质量提升了很多也学到了很多消息队列设计和代码重构的经验。

代码重构是一个程序员的基本功也是提升代码质量和可维护性的重要手段。好的代码不仅能运行还要易读易维护易扩展这需要我们不断重构优化追求代码的优雅。

消息队列是分布式系统中很重要的组件用好消息队列能提升系统的性能和可靠性但是也要注意代码的设计和最佳实践避免踩坑。

希望我的这些经历和经验能对大家有所帮助特别是做后端开发和消息队列的朋友。

最后用一句话结尾：

"代码是写给人看的顺便能在机器上运行。追求代码的优雅是每个程序员的必修课。"

愿大家都能写出优雅的代码做出稳定可靠的系统。