消息抽象层设计和实现-OSS.DataFlow

  前面已经介绍了消息生产消费中间类库(OSS.DataFlow)的简单使用,这篇主要介绍内部的设计实现。主要内容包含:

  1. 消息生产消费的抽象设计。

  2. 具体使用示例

一. 消息生产消费的抽象设计。

  需要首先强调的是,这里的生产消费抽象主要在业务使用层面,抛开具体的RabbitMQ之类的消息队列产品。可能说起来比较模糊,我们先看下常见的业务调用消息队列的情况:

  上边是一个常见的通过消息队列将业务异步拆解的模型,按道理结构已经十分简单了,特别是对于一个相对稳定的业务,代码基本变动不大的情况下,这个模型基本足够了。不过对于一个爱折腾的开发人,不玩点花里胡哨的,存在感就没了,经过各种假设论证之后,终于找到了这么几个不满意的地方:

 

  1.  开发,测试,生产,都需要搭建对应的Rabbit实例,特别是开发测试环境抢消息,再加上多人开发,大麻烦可能没有,但小麻烦一定是不断的。

  2.  业务代码中,直接调用了具体消息队列的产品,当某一个模块消息量快速上升,无法局部切换队列产品。(当然你也可以切分出独立的服务,但是耗时,代价相对较大)。

  3.  对于同一解决方案内的异步消息,或多或少的会出现生产消息调用和消费消息调用代码分散(比如用户登录后添加日志)。

  4. 当前的项目代码依赖外部消息队列或者数据库(如果公司偏项目型,新项目无法轻装上阵)

 

  以上臆测仅个人偏见,仅供参考。在我的角度,我希望在业务调用消息中转的过程中,需要面向的是接口,在需要的时候适配即可,所以我尝试添加一个轻简的中间层。

  这个中间层第一件事就是隔离,草图设计如下:

  通过全局 DataFlowFactory 能够创建消息发布者(IDataPublisher),并能够注入订阅者(IDataSubscriber) ,业务层只需要通过 IDataPublisher,IDataSubscriber 接口交互,和具体消息存储设施脱离。顺着这个思路,当业务需要生产写入消息时,创建发布者,并通过发布者写入消息,并完成订阅者的回调,这个环路即可完成。

  现在只需要解决两个问题:

  1. 创建的发布者,如何实现不同场景的扩展。

  2.  如何完成对应的订阅者(支持同一消息类型,多个订阅者)的回调。

  这里我引入一个全局管理对象(DataFlowManager),内部的调用过程图示如下:

 

  在 DataFlowManager 中提供了  PublisherProvider 公开属性,可用来扩展不同消息设施的发布者实现。同时,提供了 NotifySubscriber 方法,作为已注册消息订阅者的统一触发入口(内部完成了多个订阅者的调度,当然如果针对特殊消费者调用,用户也可以跳过注册订阅者,自由实现订阅处理)。

  通过上边的整个过程,完全实现了消息中间层的功能,以插件的形式将具体的消息设施在程序的全局入口注入,这样就可以针对不同环境不同业务模块(入口的参数 source_name 控制)做定制化。同时,在大多数的项目中(包括在开发环境中)并无需立即使用独立的消息设施,所以在中间件的内部,提供了一个默认的内存消息队列实现,这样也保证了类库的即引即用,扩展后的图示如下:

  根据上边的过程图可以看出,内部的默认队列,和外部队列所处统一层级,当没有提供用户自定义 PublisherProvider时(或者Provider 返回的 DataPublisher为空),系统会执行内部的默认队列实现。

二. 具体使用示例

  上边展示了内部设计,这里介绍具体的代码使用层面,看在实际的使用中是如何简化内敛整个消息的处理。

  1.  业务侧调用:

   上边演示了消息Key为 “ P-S-Msg ” ,有两个订阅者(这里使用了委托方法,也可以传入继承IDataSubscriber<MsgData> 接口的实例),并且创建了一个名为 _publisher 的发布者(如果没有注册其他消息存储适配实现,会走默认内部消息队列实现,即创建一个名为 NewSource 的队列)。除了上边的使用方式,有些时候我们的生产和消费代码都是同一个服务内部,比如用户登录和添加登录日志,这个时候提供了一个更简单的方式:

   可以看出,具体的业务代码相对很清晰,不需要关注具体的消息底层实现,或者什么触发方式(定时,webhook或者内部消费线程)。而这些内容全部转移到全局模块插件化适配。

二. 消息的底层适配:

 

   上边的代码,自定义了一个 CustomMsgStorage 消息存储适配器,并在全局初始化时,赋值给  DataFlowManager.PublisherProvider ,在这个适配器里,约定了当  source_name 等于 CustomStorageQueue 时返回消息发布者 CustomMsgStoragePublisher,当然这个具体实现可以替换成 RabbitMQ,Redis,Mysql 等等,当消息实际消费触发时,调用 DataFlowManager.NotifySubscriber() 方法即可,在上边的测试用例里,我简化了这个过程,直接调用,实际场景根据情况调整即可(比如放在RabbitMQ的消费监控线程,或者读取Mysql数据的定时任务中)。

  简单来说,DataFlowFactory处理业务使用接口(通过 msgkey 关联生产者和消费订阅者),DataFlowManager 控制具体的消息适配(通过 source_name 来控制底层适配), 基本解决了前面我所顾虑的问题。

如果你已经看到这里,并且感觉还行的话可以在下方点个赞,或者也可以关注我的公总号(见二维码)