整合MediatR&Cap实现EventBus

标签：MediatR Cap name contentObj 事件 EventBus public string

在软件开发中，事件早已被我们所熟悉，一个按钮按下，产生中断事件，一个回车，前端页面有侦听事件，在事件风暴建模活动中，事件也是作为领域建模的突破口，事件的重要性不言而喻。其本质是发生的事实到引发了相关事情，在这其中的传递的信息便是事件的内容。就如同猫叫了，引发着老鼠跑了，主人醒了，其中的事件便是猫叫了，而该事件是猫执行叫的动作后的结果。

领域事件

在领域驱动设计中，最开始的版本中并没有领域事件的概念，在 DDD 社区对领域驱动设计的内容不断的充实中，引入了领域事件。领域事件的命名遵循英语中的“名词 + 动词过去分词”格式，如，提交订单后发布的 OrderCreated 事件，订单完成后 OrderCompleted 事件，用以表示我们建模的领域中发生过的一件事情，也符合着事件本身是具有时间特征。对于事件所通知到的范围的不同，领域事件本身也分为两类：

领域内事件，即领域事件，作用于进程内的通信。
领域间事件，即集成事件，作用于进程间的通信。

为实现事件的发布，可以借助一些现有的包集成到系统中，但又为了隔离使用上依赖了具体的技术实现，我们可以对其进行一定的封装。

MediatR

MediatR是.NET中的开源简单中介者模式实现，它通过一种进程内消息传递机制（无其他外部依赖），进行请求/响应、命令、查询、通知和事件的消息传递，并通过泛型来支持消息的智能调度。

https://github.com/jbogard/MediatR

Cap

CAP 本身就是一个EventBus，能处理进程内消息，同时也是一个在微服务或者SOA系统中解决分布式事务问题的一个框架。它有助于创建可扩展，可靠并且易于更改的微服务系统。

https://github.com/dotnetcore/CAP

抽象EventBus

新建EventBus.Abstractions类库，用来抽象领域事件和集成事件的发布。

dotnet new classlib --name EventBusDemo.EventBus.Abstractions

在MediatR中，有一个抽象的Contracts类库，方便我们构建一个抽象的EventBus。

MediatR发布事件时，事件类上必须要有标记，比如如下INotification标记，因此，此处增加领域事件接口约束，来隔离INotification。

using MediatR;

namespace EventBusDemo.EventBus.Abstractions.Local;

public interface IDomainEvent : INotification
{
}

本地事件EventBus, 参照MediatR中已有接口，替换约束类。

namespace EventBusDemo.EventBus.Abstractions.Local;

public interface ILocalEventBUs
{
    Task Publish(IDomainEvent domainEvent, CancellationToken cancellationToken = default);

    Task Publish<T>(T domainEvent, CancellationToken cancellationToken = default)
        where T : IDomainEvent;
}

同样，增加集成事件，Cap框架没有限制集成事件类的约束条件，而为了能够统一标识项目中的集成事件，还是有必要设置一个接口约束。

namespace EventBusDemo.EventBus.Abstractions.Distributed;

public interface IIntegrationEvent
{
}

增加集成事件的EventBus，参照Cap框架中发布时的接口并用IIntegrationEvent来约束事件。

namespace EventBusDemo.EventBus.Abstractions.Distributed;

public interface IDistrbutedEventBus
{
    Task PublishAsync<T>(string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default)
        where T : IIntegrationEvent;

    Task PublishAsync<T>(string name, T? contentObj, IDictionary<string, string?> headers, CancellationToken cancellationToken = default)
        where T : IIntegrationEvent;

    void Publish<T>(string name, T? contentObj, string? callbackName = null)
        where T : IIntegrationEvent;

    void Publish<T>(string name, T? contentObj, IDictionary<string, string?> headers)
        where T : IIntegrationEvent;

    Task PublishDelayAsync<T>(TimeSpan delayTime, string name, T? contentObj, IDictionary<string, string?> headers, CancellationToken cancellationToken = default)
        where T : IIntegrationEvent;

    Task PublishDelayAsync<T>(TimeSpan delayTime, string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default)
        where T : IIntegrationEvent;

    void PublishDelay<T>(TimeSpan delayTime, string name, T? contentObj, IDictionary<string, string?> headers)
        where T : IIntegrationEvent;

    void PublishDelay<T>(TimeSpan delayTime, string name, T? contentObj, string? callbackName = null)
        where T : IIntegrationEvent;
}

实现EventBus

新建一个EventBus类库，来集成Cap和MediatR并实现EventBus.Abstractions。

dotnet new classlib --name EventBusDemo.EventBus

增加MediatR和Cap的包，Cap需要的具体存储和传输，可以在启动项目中配置。

<ItemGroup>
  <PackageReference Include="DotNetCore.CAP" Version="7.2.0" />
  <PackageReference Include="MediatR" Version="12.1.1" />
</ItemGroup>

<ItemGroup>
  <ProjectReference Include="..\EventBusDemo.EventBus.Abstractions\EventBusDemo.EventBus.Abstractions.csproj" />
</ItemGroup>

实现ILocalEventBus，实则时注入MediatR，将领域事件交给MediatR处理。

using EventBusDemo.EventBus.Abstractions.Local;
using MediatR;

namespace EventBusDemo.EventBus.Local;

public class LocalEventBus : ILocalEventBUs
{
    private readonly IPublisher _publisher;

    public LocalEventBus(IPublisher publisher)
    {
        _publisher = publisher;
    }

    public async Task Publish(IDomainEvent domainEvent, CancellationToken cancellationToken = default)
    {
        await _publisher.Publish(domainEvent, cancellationToken);
    }

    public async Task Publish<T>(T domainEvent, CancellationToken cancellationToken = default) where T : IDomainEvent
    {
        await _publisher.Publish(domainEvent, cancellationToken);
    }
}

实现IDistributedEventBus，也是注入Cap，将集成事件交给Cap处理。

using DotNetCore.CAP;
using EventBusDemo.EventBus.Abstractions.Distributed;

namespace EventBusDemo.EventBus.Distributed
{
    public class DistributedEventBus : IDistrbutedEventBus
    {
        private readonly ICapPublisher _capPublisher;

        public DistributedEventBus(ICapPublisher capPublisher)
        {
            _capPublisher = capPublisher;
        }

        public void Publish<T>(string name, T? contentObj, string? callbackName = null) where T : IIntegrationEvent
        {
            _capPublisher.Publish(name, contentObj, callbackName);
        }

        public void Publish<T>(string name, T? contentObj, IDictionary<string, string?> headers) where T : IIntegrationEvent
        {
            _capPublisher.Publish(name, contentObj, headers);
        }

        public async Task PublishAsync<T>(string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default) where T : IIntegrationEvent
        {
            await _capPublisher.PublishAsync(name, contentObj, callbackName, cancellationToken);
        }

        public async Task PublishAsync<T>(string name, T? contentObj, IDictionary<string, string?> headers, CancellationToken cancellationToken = default) where T : IIntegrationEvent
        {
            await _capPublisher.PublishAsync(name, contentObj, headers, cancellationToken);
        }

        public void PublishDelay<T>(TimeSpan delayTime, string name, T? contentObj, IDictionary<string, string?> headers) where T : IIntegrationEvent
        {
            _capPublisher.PublishDelay(delayTime, name, contentObj, headers);
        }

        public void PublishDelay<T>(TimeSpan delayTime, string name, T? contentObj, string? callbackName = null) where T : IIntegrationEvent
        {
            _capPublisher.PublishDelay(delayTime, name, contentObj, callbackName);
        }

        public async Task PublishDelayAsync<T>(TimeSpan delayTime, string name, T? contentObj, IDictionary<string, string?> headers, CancellationToken cancellationToken = default) where T : IIntegrationEvent
        {
            await _capPublisher.PublishDelayAsync(delayTime, name, contentObj, headers, cancellationToken);
        }

        public async Task PublishDelayAsync<T>(TimeSpan delayTime, string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default) where T : IIntegrationEvent
        {
            await _capPublisher.PublishDelayAsync(delayTime, name, contentObj, callbackName, cancellationToken);
        }
    }
}

如此一来，在Domain或是Application中使用IEventBus时，对底层具体使用的包便无感知(MediatR的服务注册和Cap的配置部分还可抽象一层来隔离)。

发布领域事件

如在DomainService中，将事件发布出去，通过观察者模式，找到对应的处理器，完成其他业务逻辑，以此来达到代码解耦。此处直接在服务中调用localEventBus使用，另外一种方案是在DbContext保存时通过SaveChange方法中来调用，这种情况需要事件先保存在某个位置，如聚合本身，后续文章在提到事务时一并扩展。

public class OrderService
{
    private readonly ILocalEventBus _localEventBus;

    public OrderService(ILocalEventBus localEventBus)
    {
        _localEventBus = localEventBus;
    }

    public async Task<Order> CreateOrder(string name)
    {
        var order = new Order()
        {
            Id = Guid.NewGuid(),
            Name = name,
        };

        await _localEventBus.Publish(new OrderCreatedDomainEvent()
        {
            Id = order.Id,
            Name = order.Name
        });

        return order;
    }
}

发布集成事件

如在AppService中需要通过集成事件对外通知时，注入IDistributedEventBus，便可通过底层Cap将事件广播出去，由此实现发布订阅模式。此处对于事件名还可以融合到事件本身，后续文章中扩展。

public class OrderAppService
{
    private readonly IDistributedEventBus _distributedEventBus;
    private readonly OrderService _orderService;

    public OrderAppService(
        IDistributedEventBus distributedEventBus,
        OrderService orderService)
    {
        _distributedEventBus = distributedEventBus;
        _orderService = orderService;
    }

    public async Task<Guid> CreateOrder(string name)
    {
        var order = await _orderService.CreateOrder(name);

        await _distributedEventBus.PublishAsync("OrderCreated", new OrderCreatedIntegrationEvent()
        {
            Id = order.Id,
            Name = name
        });

        return order.Id;
    }
}

2023-08-29,望技术有成后能回来看见自己的脚步

标签：MediatR,Cap,name,contentObj,事件,EventBus,public,string
From： https://www.cnblogs.com/CKExp/p/17664069.html