// 全局初始化，注入订阅者实现

    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"});    

    // 直接注册消费实现并获取消息发布接口

    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"});

　　前边介绍了业务接口的使用，和具体消息队列或数据库等隔离，这是对接业务层面的使用。因为业务场景不同，不同的项目对消息的响应速度和处理机制又各有需求，所以 OSS.DataFlow 同样提供了对接消息产品的扩展接口，方便使用者适配已有消息基础设施。

1. 消息存储适配接口

　　对于事件消息处理，需要关注两件事情：接收存储 和 消费触发。在类库中提供了 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 业务属性实现对不同业务需求返回不同的具体实现。

2. 默认实现介绍

　　借助.Net 自身的内存消息队列，在类库中提供了默认的内部消息存储转发实现（内存级别），使用者可以自行实现扩展相关接口并进行全局配置。

　　内置的.Net Core消息队列， 设置了默认1个队列，最大并发为32线程。 如果需要可以通过设置DataPublisherOption的source_name，类库将会为每个source_name 创建独立的内存队列。

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