系统重构解耦的过程涉及不同领域服务分拆,或同一服务下实时响应部分和非响应部分分拆,分解后的各部分通过异步消息的流转传递,完成整体的业务逻辑,但是频繁的在业务层面直接调用不同消息队列的SDK,不够简洁
OSS.Dataflow主要实现异步消息传递的过程抽象,在业务层面提供消息发布订阅的统一抽象接口,在业务逻辑分支之间,以简单的调用完成消息的传递,和具体的消息存储触发实现无关。
同时,在底层的存储和触发层面提取接口,能够在系统的全局适配具体的消息基础设施。在这些接口之上,实现事件处理器,通过消息的重复投放,实现事件执行的容错补充机制。
详细说明文档:https://www.yuque.com/osscore/nfbh0u/aydlz9
使用时可以通过Nuget直接安装,也可以通过命令行: Install-Package OSS.DataFlow
组件的使用非常简单,只需要关注:
a. 消息发布者接口,由组件注册时返回,供业务方法调用传入消息体。
b. 消息订阅(消费)者接口实现或委托方法,在组件注册时传入。
具体示例:
a. 消息的发布订阅独立调用示例
// 全局初始化,注入订阅者实现 const string msgPSKey = "Publisher-Subscriber-MsgKey"; DataFlowFactory.RegisterSubscriber<MsgData>(msgPSKey, async (data) => { // 当前通过注入消费的委托方法,也可通过接口实现 // DoSomething(data); return true; }); // 获取发布者接口 private static readonly IDataPublisher publisher = DataFlowFactory.CreatePublisher(); // 业务方法中发布消息 await publisher.Publish(msgPSKey,new MsgData() {name = "test"});
b. 消息的流式调用示例
// 直接注册消费实现并获取消息发布接口 private static readonly IDataPublisher _delegateFlowpusher = DataFlowFactory.RegisterFlow<MsgData>("delegate_flow",async (data) => { // 当前通过注入消费的委托方法,也可通过接口实现 // DoSomething(data); return true; }); // 业务方法中发布消息 await _delegateFlowpusher.Publish("normal_flow",new MsgData() {name = "test"});
如上,只需要获取发布者,并注入消费实现,即可完成整个消息的异步消费处理,同一个消息key可以注册多个消费实现,当有消息进入消费时,会并发处理。
前边介绍了业务接口的使用,和具体消息队列或数据库等隔离,这是对接业务层面的使用。因为业务场景不同,不同的项目对消息的响应速度和处理机制又各有需求,所以 OSS.DataFlow 同样提供了对接消息产品的扩展接口,方便使用者适配已有消息基础设施。
- 消息存储适配接口
对于事件消息处理,需要关注两件事情:接收存储 和 消费触发。在类库中提供了 DataFlowManager 消息流管理类,用户可以通过实现IDataPublisherProvider接口,完成具体的存储实现。
同时在不同的消息产品触发消费时(比如数据库定时任务或者RabbitMQ消费), 调用通知方法(NotifySubscriber ),来触发通过类库注册的具体的业务订阅处理。
// 消息流核心部件管理者 public static class DataFlowManager { /// <summary> /// 自定义 数据流发布(存储)实现的 提供者 /// </summary> public static IDataPublisherProvider PublisherProvider { get; set; } /// <summary> /// 通过自定义消息触发机制通知订阅者 /// 调用时请做异常拦截,防止脏数据导致 msgData 类型错误 /// </summary> /// <param name="msgDataKey"></param> /// <param name="msgData">消息内容,自定义触发时,请注意和注册订阅者的消费数据类型转换安全</param> /// <returns></returns> public static Task<bool> NotifySubscriber(string msgDataKey, object msgData) { .... } }
关于 IDataPublisherProvider
public interface IDataPublisherProvider { /// <summary> /// 数据发布者 /// </summary> /// <param name="option"></param> /// <returns> 返回消息发布接口实现 </returns> IDataPublisher CreatePublisher(DataPublisherOption option); } /// <summary> /// 数据的发布者 /// </summary> public interface IDataPublisher { /// <summary> /// 推进数据(存储具体消息队列或者数据库实现) /// </summary> /// <param name="dataKey"></param> /// <param name="data"></param> /// <returns>是否推入成功</returns> Task<bool> Publish<TData>(string dataKey,TData data); }
可以看到 IDataPublisher 接口负责具体的存储实现,可以根据 DataPublisherOption 的 source_name 业务属性实现对不同业务需求返回不同的具体实现。
- 默认实现介绍
借助.Net 自身的内存消息队列,在类库中提供了默认的内部消息存储转发实现(内存级别),使用者可以自行实现扩展相关接口并进行全局配置。
内置的.Net Core消息队列, 设置了默认1个队列,最大并发为32线程。 如果需要可以通过设置DataPublisherOption的source_name,类库将会为每个source_name 创建独立的内存队列。
在有些比较重要的业务处理中,如果发生异常(如网络超时)等操作,会要求过段时间后重复执行进行错误补偿,借助OSS.DataFlow 类库的消息存储和转发接口,提供了 FlowEventProcessor<TIn, TOut> 的事件处理器,在事件处理器内部完成了异常拦截重试的封装。
具体的过程就是当异常发生时,处理器通过将入参包装(FlowEventInput)通过消息流保存,具体的重试触发实现间隔,由开发者根据 MsgKey和参数 FlowEventInput 自行定制实现即可。
一. 使用示例:
// 具体执行事件 public class TestEvent:IFlowEvent<TestCount, TestCount> { public Task<TestCount> Execute(TestCount input) { input.count++; if (input.count < 10) { throw new ArgumentException("小于当前要求的条件"); } return Task.FromResult(input); } public Task Failed(TestCount input) { Console.WriteLine(input.count); return Task.CompletedTask; } } public class TestCount { public int count { get; set; } } // 单元测试方法 [TestMethod] public async Task DataStackTest() { var o = new FlowEventOption() { event_msg_key = "Test_flow_event_msg", flow_retry_times = 4, // 经过消息流重试次数 func_retry_times = 1 // 当前执行方法内部直接串联循环重试次数 }; var flowProcessor = new FlowEventProcessor<TestCount,TestCount>(new TestEvent(), o); var countPara = new TestCount() {count = 0}; var countRes = await flowProcessor.Process(countPara); Assert.IsNull(countRes); // 首次抛出异常,拦截返回空 await Task.Delay(5000); // 异步消息队列消费缓冲 // 总执行次数 = (flow_retry_times+1)*(func_retry_times+1) = (4+1)*(1+1) = 10 Assert.IsTrue(countPara.count==10);// 默认消息队列实现是内存级,countPara引用不变 }
二. 使用介绍
上例中 TestCount为默认入参和出参类型。 继承至 IFlowEvent<TestCount, TestCount> 的 TestEvent 为具体执行事件,其定义如下:
public interface IFlowEvent<in TIn,TOut> { /// <summary> /// 具体事件执行 /// </summary> /// <param name="input"></param> /// <returns></returns> Task<TOut> Execute(TIn input); /// <summary> /// 最终失败执行方法 /// </summary> /// <param name="input"></param> /// <returns></returns> Task Failed(TIn input); }
FlowEventOption为当前事件执行重试等参数项。其定义如下:
public class FlowEventOption { /// <summary> /// 保存事件消息的key(用来保存消息到消息流订阅重试) /// </summary> public string event_msg_key { get; set; } /// <summary> /// 异常时 推送消息到消息流,通过订阅方式重试运行次数,默认:0 /// </summary> public int flow_retry_times { get; set; } = 1; /// <summary> /// 异常时 在推送消息流之前,在当前执行方法内部直接串联循环重试次数,默认:1 /// </summary> public int func_retry_times { get; set; } = 0; /// <summary> /// 消息流的可选项 /// </summary> public DataFlowOption flow_option { get; set; } }