CAP事件总线在NetCore中的应用

标签：订阅 NetCore CAP 总线重试消息 message public

在前面的文章中，我们介绍过 Abp自带的本地事件总线，但它有几点不足

1：缺乏失败重试机制，即若发布事件失败或者订阅事件处理失败，他没有重试机制，导致业务和数据异常。

2：缺乏对存储的支持，没有集成数据库，对事件进行存储，事件的处理都是基于内存。

下面我们介绍CAP事件总线这个组件工具，它很好的实现了以上两个问题，同时，也是微软官方应用中推荐的事件总线组件。

CAP官方文档：https://cap.dotnetcore.xyz/user-guide/zh/getting-started/quick-start/

因内容比较多，我们今天先介绍CAP的基本应用，基于内存存储的模式。

我们新建一个web mvc的项目CAP事件总线，项目环境基于.NET6.0

Nuget中我们需要引用3个包：
DotNetCore.CAP：CAP的核心包

DotNetCore.CAP.InMemoryStorage：CAP的内存存储包

Savorboard.CAP.InMemoryMessageQueue：CAP基于内存的对列包

下面2个因我们示例采用内存存储才需要，若我们使用数据库，则选择安装对应CAP的数据库组件

然后，在Program.cs中配置CAP服务，若不是.NET6.0，则在startup.cs中注入

var services = builder.Services;
services.AddCap(x =>
{
    x.UseInMemoryStorage();
    x.UseInMemoryMessageQueue();
    x.FailedRetryInterval = 20; //重试的间隔事件，默认是60s，官网上配置有详细说明
});

完成以上配置，我们就可以开始使用CAP来发布和订阅事件了

发布消息，主要使用ICapPublisher对象

我们在Home控制器上初始化ICapPublisher 来发送一个简单的消息

消息体包含了消息名称和消息内容，消息名称用于订阅识别，如果是接入RabbitMQ，则对应它的Queue的名称

using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;

namespace CAP事件总线.Controllers
{
    public class HomeController : Controller
    {

        private readonly ICapPublisher  _capPublisher;

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

        public IActionResult Index()
        {
            //发送消息
            _capPublisher.Publish("test.message", "我发出消息啦"); 

            return View();
        }

      
    }
}

消息发送的方法支持多样化，我们看下有哪些发送方法

　　　　　　/// <summary>
        /// 异步发布对象消息
        /// </summary>
        /// <param name="name">消息名称</param>
        /// <param name="contentObj">消息内容体，可以是实体对象，也可以是匿名对象</param>
        /// <param name="callbackName">回调的消息名称，指消息发布后，由订阅端处理，若订阅端处理成功并返回内容，默认会再发布一个消息名为callbackName的消息，可以接着订阅处理</param>
        /// <param name="cancellationToken">是否取消</param>
        Task PublishAsync<T>(string name, T? contentObj, string? callbackName = null, CancellationToken cancellationToken = default);

        /// <summary>
        /// 使用自定义头异步发布对象消息
        /// </summary>
        /// <typeparam name="T">实体对象</typeparam>
        /// <param name="name">消息内容体，可以是实体对象，也可以是匿名对象</param>
        /// <param name="contentObj">可序列化的消息体</param>
        /// <param name="headers">头信息</param>
        /// <param name="cancellationToken"></param>
        Task PublishAsync<T>(string name, T? contentObj, IDictionary<string, string?> headers, CancellationToken cancellationToken = default);

        /// <summary>
        /// Publish an object message.
        /// </summary>
        /// <param name="name">the topic name or exchange router key.</param>
        /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
        /// <param name="callbackName">callback subscriber name</param>
        void Publish<T>(string name, T? contentObj, string? callbackName = null);

        /// <summary>
        /// Publish an object message.
        /// </summary>
        /// <param name="name">the topic name or exchange router key.</param>
        /// <param name="contentObj">message body content, that will be serialized. (can be null)</param>
        /// <param name="headers">message additional headers.</param>
        void Publish<T>(string name, T? contentObj, IDictionary<string, string?> headers);

总结起来呢，就是以下几点

1：必须要有一个消息名称name

2：消息体只要是object类型的都可以，实体需要可序列化

3：支持消息订阅处理完后的回调

4：支持添加消息头

上面的示例中，我们是用最简单的方式，发送一个消息名为test.message的消息，内容就是一个字符串

下面我写一个订阅，订阅有两种写法均可，一种是直接使用控制器的，另一种就是在服务类中

我们先上控制器上的代码

using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;

namespace CAP事件总线.Controllers
{
    public class ConsumerController : Controller
    {
        [NonAction]
        [CapSubscribe("test.message")]
        public void ReceiveMessage(string message)
        {
            Console.WriteLine("message  is:" + message);
        }
    }
}

所有的订阅都是采用 CapSubscribe+消息名的方式

NonAction：这个是MVC自带的特性，跟CAP无关。若将NonAction属性应用在Controller中的Action方法上，即使该Action方法是公共方法，也会告知ActionInvoker不要选取这个Action来执行。这个属性主要用来保护Controller中的特定公共方法不会发布到Web上。或是当功能尚未开发完成就要进行部署时，若暂时不想将此方法删除，也可以使用这个属性。简单的说，就是通过url访问不了，避免对外暴露。

我们运行起来看下：

从截图中，直观看到，消息订阅是执行成功了。

若订阅不是在控制器上写呢，而是在服务层方法呢，应该怎么订阅？为了验证这个，我们新建一个MessageHandler 消息处理类

using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;

namespace CAP事件总线.Models
{
    public class MessageHandler: ICapSubscribe //消息的处理
    {
        [CapSubscribe("test.message")]
        public void ReceiveMessage(string message)
        {
            Console.WriteLine("在类方法中订阅 message  is:" + message);
        }

    }
}

然后我们需要注入这个类

在 programe.cs中增加

services.AddTransient<MessageHandler>();//若是服务层已经注入，则不需要

1：类方法必须继承 ICapSubscribe 接口对象

2：类必须依赖注入到服务中

看下执行结果：

可以看到方法中的订阅成功了。

以上是一个简单的CAP基于内存存储的简单示例。

但是从运行结果来看，CAP貌似不支持同一个消息名，多个订阅者。

我们刚在控制器层有订阅，现在类方法中也定义了同一个消息的订阅，但最终它只执行了一个。这也是跟ABP有区别的地方，ABP支持多个订阅。

当然根据群里问了别人，说可以用分组来实现。下一次再试试。

下一步，我们继续验证CAP的重试机制

我们定义一个静态的 i 变量，初始化为0。每次执行订阅，都对 i+1。别问我每次执行为啥 i 不是重新初始化，因为是静态变量哈，自己理解。

在订阅端，我们设置当I<3时，都抛出异常，大于3

上代码

using DotNetCore.CAP;
using Microsoft.AspNetCore.Mvc;

namespace CAP事件总线.Models
{
    public class MessageHandler: ICapSubscribe //消息的处理
    {
        static int i = 0;
        [CapSubscribe("test.message")]
        public void ReceiveMessage(string message)
        {
            i++;
            if(i<3)
            {
                throw new Exception("i 小于3，抛出异常");
            }
            Console.WriteLine("在类方法中订阅 message  is:" + message);
        }


    }
}

看运行结果