.netcore利用DI實現訂閱者模式 - xms

結合DI，實現發布者與訂閱者的解耦，屬於本次事務的對象主體不應定義為訂閱者，因為訂閱者不應與發布者產生任何關聯

一、發布者訂閱者模式

發布者發出一個事件主題，一個或多個訂閱者接收這個事件，中間通過事件總線通訊(消息隊列)，並且發布者與訂閱者這兩者間是無狀態的，根據產品實際場景需要，可以自己實現單機單點的發布訂閱，也可選擇使用目前流行的分布式消息中間件：

RabbitMQ、ActiveMQ、RocketMQ、kafka等

二、觀察者與訂閱者的區別

觀察者與業務主體是耦合的，並且是即時通知的；訂閱者與業務主體完全解耦，只通過中間的信息通道通知，互相不知道對方的存在，可以是同步也可以是異步

三、具體實現

本文主要講解單點模式，有需要隨時可以擴展為分布式方案

發布者接口：

namespace Xms.Event.Abstractions
{
    /// <summary>
    /// 事件發布接口
    /// </summary>
    public interface IEventPublisher
    {
        /// <summary>
        /// 發布事件
        /// </summary>
        /// <typeparam name="TEvent">事件類型</typeparam>
        /// <param name="e"></param>
        void Publish<TEvent>(TEvent e);
    }
}

發布者實現：

using System;
using System.Linq;
using Xms.Event.Abstractions;
using Xms.Infrastructure.Inject;
using Xms.Logging.AppLog;

namespace Xms.Event
{
    /// <summary>
    /// 事件發布者
    /// </summary>
    public class EventPublisher : IEventPublisher
    {
        private readonly ILogService _logService;
        private readonly IServiceResolver _serviceResolver;

        public EventPublisher(ILogService logService
            , IServiceResolver serviceResolver)
        {
            _logService = logService;
            _serviceResolver = serviceResolver;
        }

        #region Methods

        /// <summary>
        /// 發布事件
        /// </summary>
        /// <typeparam name="TEvent">事件類</typeparam>
        /// <param name="e">事件對象</param>
        public virtual void Publish<TEvent>(TEvent e)
        {
            //獲取所有事件接收者
            var consumers = _serviceResolver.GetAll<IConsumer<TEvent>>().ToList();
            foreach (var consumer in consumers)
            {
                try
                {
                    //處理事件
                    consumer.HandleEvent(e);
                }
                catch (Exception exception)
                {
                    _logService.Error(exception);
                }
            }
        }

        #endregion Methods
    }
}

訂閱(消費)者接口：

namespace Xms.Event.Abstractions
{
    /// <summary>
    /// 事件接收接口
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public interface IConsumer<T>
    {
        /// <summary>
        /// 處理事件
        /// </summary>
        /// <param name="eventMessage">事件</param>
        void HandleEvent(T eventMessage);
    }
}

事件(消息)：

這里只給出一個作為示例，實際上一般會有記錄的：“創建”、“修改”、“刪除”，流程相關的：“發起審批”、“審批通過”、“審批完成”等等

namespace Xms.Flow.Core.Events
{
    /// <summary>
    /// 工作流啟動后事件
    /// </summary>
    public class WorkFlowStartedEvent
    {
        public WorkFlowStartUpContext Context { get; set; }
        public WorkFlowExecutionResult Result { get; set; }
    }
}

服務注冊：

詳細實現回看.netcore之DI批量注入(支持泛型)

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Xms.Core;
using Xms.Infrastructure.Inject;

namespace Xms.Event
{
    /// <summary>
    /// 事件模塊服務注冊
    /// </summary>
    public class ServiceRegistrar : IServiceRegistrar
    {
        public int Order => 1;

        public void Add(IServiceCollection services, IConfiguration configuration)
        {
            //event publisher
            services.AddScoped<Event.Abstractions.IEventPublisher, Event.EventPublisher>();
            //event consumers
            services.RegisterScope(typeof(Event.Abstractions.IConsumer<>));
        }
    }
}

四、應用場景

比如在工作流啟動審批后發送通知

using System.Collections.Generic;
using Xms.Context;
using Xms.Event.Abstractions;
using Xms.Flow.Core.Events;
using Xms.Infrastructure.Utility;
using Xms.Localization.Abstractions;
using Xms.Notify.Abstractions;
using Xms.Notify.Internal;

namespace Xms.EventConsumers.Notify
{
    /// <summary>
    /// 工作流啟動審批后發送通知
    /// </summary>
    public class WorkflowStartedNotify : IConsumer<WorkFlowStartedEvent>
    {
        private readonly IAppContext _appContext;
        private readonly ILocalizedTextProvider _loc;
        private readonly IEnumerable<INotify> _notifies;

        public WorkflowStartedNotify(IAppContext appContext
            , IEnumerable<INotify> notifies)
        {
            _appContext = appContext;
            _loc = _appContext.GetFeature<ILocalizedTextProvider>();
            _notifies = notifies;
        }
        public void HandleEvent(WorkFlowStartedEvent eventMessage)
        {
            //當前節點處理人
            foreach (var handlerId in eventMessage.Result.NextHandlerId)
            {
                //通知方式：微信、短信、郵件、系統消息等
                var msg = _loc["workflow_newtasknotify"].FormatWith(eventMessage.Context.EntityMetaData.LocalizedName);
                //發送消息
                foreach (var notifier in _notifies)
                {
                    notifier.Send(new InternalNotifyBody()
                    {
                        TypeCode = 2
                        ,
                        Subject = msg
                        ,
                        Content = "到你審批了，快到碗里來"
                        ,
                        ToUserId = handlerId
                        ,
                        LinkTo = "/entity/create?entityid=" + eventMessage.Context.EntityMetaData.EntityId + "&recordid=" + eventMessage.Context.ObjectId
                    });
                }
            }
        }
    }
}

 

四、總結

前面講解了訂閱者模式的基本概念及與觀察者的區別，后面展示了具體實現及實際應用場景，大家記住一點就行，這些設計模式最終都是為了達到解藕的目的，要查看完整代碼，請回到這一章

xms跨平台基礎框架 - 基於.netcore