后台消息队列处理简易框架

Posted 2021-04-26 腾讯课堂Coding学院

来源 /腾讯课堂Coding学院（ID：ke_coding）

导语

在很多系统架构和应用中，消息队列都是必不可缺的组件之一，它能够将系统解耦，且有扩展性强，异步通信等优点。比较适合实时性要求较低的，且比较耗时的场景。

01存在的问题

自选股后台就大量使用了分布式消息队列来提高接口层的访问效率，包括数据库写入、提醒发送等业务。但是当前后台消息队列处理程序存在以下几个问题：

代码库封装不够，有大量冗余代码。除了关注业务，需要处理太多细节问题.

例如在进行数据库操作时还需要关心通过名字服务获取数据库ip/port信息、数据库连接、可能出错的处理、log的记录等问题，导致业务代码中有大量此类相似逻辑，造成代码可读性较差，维护困难。同样在redis处理以及消息队列的处理中也有此类问题。　

代码不规范，可读性差.

代码风格引人而异，一个逻辑多种写法，条理性不够强，最主要的处理逻辑需要从main函数一步一步去查；由于没有规范，如果出现问题，想启动应用去调试，需要去程序中查找消息来源，消息解析等，调试较困难。

log日志不规范，调试日志花时间。

我认为log是一个程序最不可或缺的存在，清晰的log可以直接定位到问题出错的原因、出错的位置，让我们一眼找到解决的方法。但是由于程序中需要处理大量细节问题，并且log日志不规范，导致log很难定位到确切位置，而且调试log也需要大量时间，成本很高。

 大量相同逻辑，可维护性差。

后台消息队列的逻辑其实很简单，只需要获取消息、解析消息、处理消息，每份代码中都有类似的逻辑，完全可以提取出来，统一处理。

02主题目标

要对程序改进，首先需要梳理一下我们要解决的问题，问题也很简单：程序首先需要从队列中获取消息，然后对消息体进行解析，通过消息获取对应的处理逻辑方法，然后具体的业务逻辑是在处理方法中处理，同时还需要获取各种资源，如db、redis、queue等，考虑消息获取或者解析失败的处理、资源获取失败的处理，log的记录等问题，此外业务中还有定时消息处理的逻辑。

03解决方案

框架:

注释：

代码列表：baseObj：基类，作为扩展使用。

queueProcessor：为消息处理模块，完成消息获取、消息解析、消息处理的所有工作。

processor：从env中获取配置，定制化queueprocessor。
timeProcessor：定时处理模块，单开进程定时处理，queueprocessor通过队列queue向timeProcessor提交任务，到达时间时，timeProcessor调用queueProcessor进行处理。
service：业务逻辑处理，这是每个业务逻辑实际处理的位置，一般主要负责入库、或写入其他服务队列。
resLoader: 里面封装了beanstalk、db、rabbitmq、redis的资源获取以及相应命令执行函数。
parser：消息队列的消息解析函数，一般常用的在此模块中，也可定制。
config：配置获取函数，主要包含getRedisConf、getDbConf、getMqUrl、get，获取配置文件中相应的配置。
utils：为常用函数。

使用示例：msg的格式为：env为环境配置信息，其中configfile保存有db、queue、redis等的资源配置信息；

queuename为消息队列获取的来源（前缀beans代表是beanstalk消息队列），其配置信息也是存储在configfile中；

queuekey为消息队列的key；

msgParser为消息解析函数，可以使用parser中的，也可定制；

 acthandlers为消息处理函数，通过act来进行路由。

04优点

可配置性

1、获取消息队列可定制，rabbitmq，beanstalk支持，队列类型容易扩展，只需要在resloader中加入相应的资源获取类即可。

2、消息格式解析可定制，在parser中有常用解析，也可以自己定制，只需要在msgParser进行配置，即可方便进行消息解析。

3、消息队列处理函数通过act配置，这里是用户需要开发的服务模块。只需要写处理逻辑。

资源类封装更完善

1、扩展性强，如果有新资源需要获取，只需要在resloaders目录添加相应的加载类，即可轻松扩展。

2、对比之前的用法，框架中使用资源类更加方便，并且如果配置加载出错或者执行sql出错都会log日志记录，另外每个资源都使用单例方式，只会加载一次。如:3、log日志统一且全面，可以方便的进行分析。

4、便于调试，之前想调试程序，可能需要别人配合，或者需要自己写个生成原始消息的脚本，麻烦，框架中考虑这一点，只需几步。

05建议和思考

1、一个系统只部署一个框架，方便进行升级〔增加监控状态模块，增加cron任务管理模块〕。

2、框架的好处是可以积累，比如我们程序中获取一个db需要经过很多步，每个程序里面都有这个过程，而且每个人喜好不同，代码风格不同，多种不同的处理方式。如果有框架规范，我们只保留公认最好的。轮子可以重新造，但也要越造越好。

3、代码规范3.1   写一个消息处理，在service下开一个目录，执行程序（主程序）为目录名

3.2   调用数据库的处理建议一个py一个库，比如需要获取用户的liveid，表是livevip，库是talkdb，建议文件名为talkdb，里面定义一个类TalkDb。redis也类似处理。

3.3 处理方法如果逻辑复杂可以分成几个文件来定义，如果有很多逻辑很小，可以一个文件处理。

3.4 定义一个环境配置文件：env.py，进行各种配置。

06后续改进

1. 增加状态监控模块，可以实时查看程序运行情况。当前自选股监控系统可以使我们被动警告，但是很多时候，我们想去主动获取任务运行情况，监控系统可以使我们主动获取消息实时处理过程，利于问题排查。

2. 增加后台部署发现机制。当前后台处理脚本部署较为分散，可以增加发现模块进行管理。

业界最顶尖的技术大咖/最权威的实战分享/最前沿的行业资讯/尽在腾讯课堂Coding学院

长按二维码关注Coding学院