CAP 　　　　　　　　　　　　　　　　　　　　英文

CAP 是一个基于 .NET Standard 的库，它是一个处理分布式事务的解决方案，具有 EventBus 的功能，轻量、易用且高效。

在构建 SOA 或微服务系统时，我们通常需要使用事件来集成各个服务。在这个过程中，简单使用消息队列并不能保证可靠性。CAP 采用本地消息表与当前数据库集成的方案来解决分布式系统调用过程中可能出现的异常情况。它可以确保在任何情况下事件消息都不会丢失。

你也可以将 CAP 用作 EventBus。CAP 提供了一种更简单的方式来实现事件的发布和订阅。在订阅和发送过程中，你不需要继承或实现任何接口。

架构概述

CAP 实现了 eShop 电子书 中描述的 Outbox 模式。

入门

NuGet

你可以使用以下命令在项目中安装 CAP。

PM> Install-Package DotNetCore.CAP

CAP 支持大多数流行的消息队列作为传输，以下包可供安装：

PM> Install-Package DotNetCore.CAP.Kafka
PM> Install-Package DotNetCore.CAP.RabbitMQ
PM> Install-Package DotNetCore.CAP.AzureServiceBus
PM> Install-Package DotNetCore.CAP.AmazonSQS
PM> Install-Package DotNetCore.CAP.NATS
PM> Install-Package DotNetCore.CAP.RedisStreams
PM> Install-Package DotNetCore.CAP.Pulsar

CAP 支持大多数流行的数据库作为事件存储，以下包可供安装：

// 选择你正在使用的数据库提供程序，事件日志表将集成到其中。

PM> Install-Package DotNetCore.CAP.SqlServer
PM> Install-Package DotNetCore.CAP.MySql
PM> Install-Package DotNetCore.CAP.PostgreSql
PM> Install-Package DotNetCore.CAP.MongoDB     //需要 MongoDB 4.0+ 集群

配置

首先，你需要在 Startup.cs 中配置 CAP：

public void ConfigureServices(IServiceCollection services)
{
    //......

    services.AddDbContext<AppDbContext>(); //可选，如果你使用 EF 作为 ORM
    services.AddSingleton<IMongoClient>(new MongoClient("")); //可选，如果你使用 MongoDB

    services.AddCap(x =>
    {
        // 如果你使用 EF，你需要添加如下配置：
        x.UseEntityFramework<AppDbContext>(); //可选，注意：你不需要再配置 x.UseSqlServer("")！CAP 可以自动发现。

        // 如果你使用 ADO.NET，选择添加你需要的配置：
        x.UseSqlServer("你的连接字符串");
        x.UseMySql("你的连接字符串");
        x.UsePostgreSql("你的连接字符串");

        // 如果你使用 MongoDB，你需要添加如下配置：
        x.UseMongoDB("你的连接字符串");  //MongoDB 4.0+ 集群

        // CAP 支持 RabbitMQ、Kafka、AzureService 作为 MQ，选择添加你需要的配置：
        x.UseRabbitMQ("主机名");
        x.UseKafka("连接字符串");
        x.UseAzureServiceBus("连接字符串");
        x.UseAmazonSQS();
    });
}

发布

在你的 Controller 中注入 ICapPublisher，然后使用 ICapPublisher 发送消息。

7.0+ 版本支持发布延迟消息。

public class PublishController : Controller
{
    private readonly ICapPublisher _capBus;

    public PublishController(ICapPublisher capPublisher)
    {
        _capBus = capPublisher;
    }

    [Route("~/adonet/transaction")]
    public IActionResult AdonetWithTransaction()
    {
        using (var connection = new MySqlConnection(ConnectionString))
        {
            using (var transaction = connection.BeginTransaction(_capBus, autoCommit: true))
            {
                //你的业务逻辑代码

                _capBus.Publish("xxx.services.show.time", DateTime.Now);

                // 发布延迟消息
                _capBus.PublishDelayAsync(TimeSpan.FromSeconds(delaySeconds), "xxx.services.show.time", DateTime.Now);
            }
        }

        return Ok();
    }

    [Route("~/ef/transaction")]
    public IActionResult EntityFrameworkWithTransaction([FromServices]AppDbContext dbContext)
    {
        using (var trans = dbContext.Database.BeginTransaction(_capBus, autoCommit: true))
        {
            //你的业务逻辑代码

            _capBus.Publish("xxx.services.show.time", DateTime.Now);
        }

        return Ok();
    }
}

订阅

在 Controller Action 中

在 Action 上添加 [CapSubscribe()] 特性来订阅消息：

public class PublishController : Controller
{
    [CapSubscribe("xxx.services.show.time")]
    public void CheckReceivedMessage(DateTime datetime)
    {
        Console.WriteLine(datetime);
    }
}

在业务逻辑服务中

如果你的订阅方法不在 Controller 中，那么你的订阅类需要实现 ICapSubscribe 接口：

namespace BusinessCode.Service
{
    public interface ISubscriberService
    {
        void CheckReceivedMessage(DateTime datetime);
    }

    public class SubscriberService: ISubscriberService, ICapSubscribe
    {
        [CapSubscribe("xxx.services.show.time")]
        public void CheckReceivedMessage(DateTime datetime)
        {
        }
    }
}

然后在 Startup.cs 中注册你实现了 ISubscriberService 的类

public void ConfigureServices(IServiceCollection services)
{
    services.AddTransient<ISubscriberService,SubscriberService>();

    services.AddCap(x=>
    {
        //...
    });
}

异步订阅

你可以实现异步订阅。订阅方法应该返回 Task 并接收 CancellationToken 作为参数。

public class AsyncSubscriber : ICapSubscribe
{
    [CapSubscribe("name")]
    public async Task ProcessAsync(Message message, CancellationToken cancellationToken)
    {
        await SomeOperationAsync(message, cancellationToken);
    }
}

使用部分主题订阅

要在类级别上对主题订阅进行分组，你可以在方法上定义部分订阅。消息队列上的订阅将是类上定义的主题和方法上定义的主题的组合。在以下示例中，当接收到 customers.create 主题的消息时，将调用 Create(..) 函数

[CapSubscribe("customers")]
public class CustomersSubscriberService : ICapSubscribe
{
    [CapSubscribe("create", isPartial: true)]
    public void Create(Customer customer)
    {
    }
}

订阅组

订阅组的概念类似于 Kafka 中的消费者组。它与消息队列中的广播模式相同，用于在多个不同的微服务实例之间处理相同的消息。

当 CAP 启动时，它将使用当前程序集名称作为默认组名。如果多个相同组的订阅者订阅了相同的主题名称，则只有一个订阅者能接收到消息。 相反，如果订阅者在不同的组中，它们都将接收到消息。

在同一应用程序中，你可以指定 Group 属性将订阅保持在不同的订阅组中：

[CapSubscribe("xxx.services.show.time", Group = "group1" )]
public void ShowTime1(DateTime datetime)
{
}

[CapSubscribe("xxx.services.show.time", Group = "group2")]
public void ShowTime2(DateTime datetime)
{
}

ShowTime1 和 ShowTime2 将会被依次调用，因为所有接收到的消息都是线性处理的。 你可以通过将 UseDispatchingPerGroup 设置为 true 来改变这种行为。

顺便说一下，你可以在配置中指定默认的组名：

services.AddCap(x =>
{
    x.DefaultGroup = "default-group-name";  
});

控制台

CAP 还提供了控制台页面，你可以轻松查看已发送和接收的消息。此外，你还可以在控制台中实时查看消息状态。使用以下命令在你的项目中安装控制台。

PM> Install-Package DotNetCore.CAP.Dashboard

在分布式环境中，控制台内置集成了 Consul 作为节点发现，同时实现了网关代理功能，你可以轻松查看节点或其他节点的数据，就像访问本地资源一样。

查看 Consul 配置文档

如果你的服务部署在 Kubernetes 中，请使用我们的 Kubernetes 发现包。

PM> Install-Package DotNetCore.CAP.Dashboard.K8s

查看 Kubernetes 配置文档

控制台的默认地址是：http://localhost:xxx/cap，你可以通过 x.UseDashboard(opt =>{ opt.MatchPath="/mycap"; }) 配置相对路径 /cap。

贡献

参与贡献的最简单方式之一是参与讨论和讨论问题。你也可以通过提交包含代码更改的拉取请求来贡献。

许可证

MIT