一文带你了解CAP的全部特性，你学会了吗？-数字星空

前言
消息发布
事务消息
消息处理
消费者特性
控制与集成特性
监控，诊断，可观察性
总结

前言

这是一篇关于 CAP 全部特性的一览文章，里面也包含了官方文档中未提及的一些特性，来看看是否有你不知道的特性，同时也是对于新用户在选型时是一个很好的参考。

CAP现在已经不仅仅是一个分布式事务的解决方案，其在消息处理方面提供的大量特性可以在面临复杂的业务场景时也游刃有余，并仍然保持轻量级、高性能、易使用的特点。

在数据兼容性方面也做了大量努力，5年前发布的 3.0 版本在不经过任何改动的情况下也可直升最新版本。同时过去7年的不断改进与完善，在稳定性方面也获得了用户的大量好评。

项目简介：

CAP 是一个处理分布式事务问题的开源解决方案，该项目诞生于2016年，目前在 Github 已经有超过 6500+ Star 和 110+ 贡献者，以及在 NuGet超 800 万的下载量。

消息发布

携带消息头

消息头可以用来传递与消息相关的元数据信息，CAP支持在消息中携带自定义头部信息，这些头部信息可以在消费者端进行读取和处理。在发布消息时添加头部信息：

var headers = new Dictionary<string, string>
{
   { "my.header.first", "first" },
   { "my.header.second", "second" }
};
await capBus.PublishAsync("test.show.time", DateTime.Now, headers);

设置消息前缀

使用前缀可以有效管理和组织消息，使得不同类型或不同应用的Topic具有一定的层次结构，CAP允许为消息设置特定的前缀，便于分类和识别。可以在配置中设置：

services.AddCap(x =>
{
   x.TopicNamePrefix = "prefix";
});

原生支持的延迟消息

延迟消息可以用于实现定时任务调度执行，也可以通过发送延迟消息来实现重试等机制，同时也在缓存失效、订单超时处理、流量削峰、异步工作流等场景有多种应用。CAP原生支持延迟消息而不需要借助消息队列的能力，例如，延迟100秒发布消息：

await capBus.PublishDelay(TimeSpan.FromSeconds(100), "test.show.time", DateTime.Now);

并行发布消息

如果你在消息发送的时候追求高性能而不是特别在意消息顺序，CAP支持开启并行发布消息配置，在内部CAP会按照批次来处理数据，这会显著提高发布速度。

services.AddCap(x =>
{
   x.EnablePublishParallelSend = true;
});

事务消息

事务消息是CAP的第一大特性，在构建 SOA 或者微服务系统的过程中，通常需要使用事件来对各个服务进行集成，在这过程中简单的使用消息队列并不能保证数据的最终一致性， CAP 采用的是和当前数据库集成的本地消息表的方案来解决在分布式系统互相调用的各个环节可能出现的异常，它能够保证任何情况下事件消息都是不会丢失的。

背景知识：https://www.cnblogs.com/savorboard/p/distributed-system-transaction-consistency.html

事务消息发送

发送事务消息需要同时确保业务消息和事件消息包含着一个数据库事务上下文中，这样才能保证数据一致性。以下是和本地事务集成的示例：

using (var connection = new MySqlConnection(ConnectionString))
{
	using (var transaction = connection.BeginTransaction(_capBus, autoCommit: true))
	{
		connection.Execute("insert into test(name) values('test’)",
			 transaction: (IDbTransaction)transaction.DbTransaction);
		await cap.PublishAsync("order.insert", DateTime.Now);
	}
}

如果你使用的第三方ORM，参考下面的链接。

FreeSQL集成示例：https://github.com/dotnetcore/FreeSql/discussions/1202
SqlSugar集成示例：https://github.com/DotNetNext/SqlSugar/issues/1207
Chloe集成示例：https://github.com/shuxinqin/Chloe/issues/328

事务消息消费

消费端事务其实没有多大必要，因为消息已经不会丢了，但是如果你想开启，也可以。消费端自动事务主要利用 CAP 提供的过滤器来进行开启。

1、创建过滤器

public class MyCapFilter : SubscribeFilter {
    private readonly AppDbContext _dbContext;
    private IDbContextTransaction _transaction;

    public MyCapFilter(AppDbContext dbContext){
        _dbContext = dbContext;
    }
    public override void OnSubscribeExecuting(ExecutingContext context){
        _transaction = _dbContext.Database.BeginTransaction();
    }
    public override void OnSubscribeExecuted(ExecutedContext context){
        _transaction.Commit();
    }
    public override void OnSubscribeException(DotNetCore.CAP.Filter.ExceptionContext context){
       _transaction.Rollback();
    }
}

2、配置过滤器

services.AddCap(opt =>
{
    // ***
}.AddSubscribeFilter<MyCapFilter>();

事务补偿

在进行多服务的业务事务操作时，可能会遇到下游执行失败的情况，CAP支持提供操作失败后的补偿机制，可以通过回调机制实现。例如：

await cap.PublishAsync("place.order", ...,"place.order.qty.deducted");

[CapSubscribe("place.order.qty.deducted")]
public object DeductProductQty(JsonElement param)
{
   var orderId = param.GetProperty("OrderId").GetInt32();
   // 业务逻辑
   return new { OrderId = orderId, IsSuccess = true };
}

消息处理

序列化

默认情况下CAP使用 Json 来序列化消息，Json现在基本已成为传输的标准格式，这这在大部分情况下工作的很好。如果你对性能、兼容性、安全性等有其他考虑，可以通过 ISerializer 接口来自定义实现支持诸如 XML、GRPC、MessagePack 等格式。

序列化 Json 默认使用 .NET System.Text.Json 实现，如果你想更换为 Newtonsoft.Json 查看这里。

services.AddSingleton<ISerializer, YourSerializer>();

过滤器

CAP 过滤器和 ASP.NET MVC 中的过滤器类似，可以在消费者被执行前、执行后、执行异常时进行拦截并处理。

public class MyCapFilter: SubscribeFilter {
    public override Task OnSubscribeExecutingAsync(ExecutingContext context){
        // 订阅方法执行前
    }
    public override Task OnSubscribeExecutedAsync(ExecutedContext context){
        // 订阅方法执行后
    }
    public override Task OnSubscribeExceptionAsync(ExceptionContext context){
        // 订阅方法执行异常
    }
}

services.AddCap(opt =>
{
    // ***
}.AddSubscribeFilter<MyCapFilter>();

消息重试

当消息发送或执行失败时，CAP支持重试机制。可以配置重试间隔和重试次数：

services.AddCap(x =>
{
   x.FailedRetryCount = 5;
   x.FailedRetryInterval = 60;
});

重试回退

重试器会定期扫描失败的消息并重新尝试，对于失败的消息默认查询当前时间回退 4 分钟前的，这是为了保证新消息不会被重试器拾取。
你可以通过 FallbackWindowLookbackSeconds 配置项来自定义配置此值。

多线程处理

多消费线程支持
多执行线程支持

自动恢复/重连

对于消息队列的自动重连功能在分布式系统和消息驱动的架构中具有重要作用，特别是在提高系统的可用性、可靠性和容错能力方面。CAP原生实现了对于连接的健康检查及自动重连功能而不是借助于客户端的实现，从而可以最大程度的保证连接的稳定性，提高系统的容错性，使得消费者能够在遇到问题时自动恢复，而不需要重启系统或应用。

这是一项内置功能，不需要任何配置。

分布式存储锁

重试线程默认每隔1分钟来读取存储的消息用于对发送或消费失败的消息进行重试，单个实例没有什么问题，在启用多个实例的场景下会有一定几率出现并发读的情况，CAP支持配置基于数据库的分布式锁，这样可以避免多实例并发读的问题，并且对异常场景的处理也进行了考虑。

services.AddCap(x =>
{
   x.UseStorageLock = true;
});

消息版本隔离

版本隔离就是利用版本特性将消息对象按照版本划分，从而来隔离不同版本的业务或实例。CAP支持版本隔离特性，可以用于多个业务场景，例如业务快速迭代，需要向前兼容、多个版本的服务端接口、不同实例使用相同的表、开发阶段不同开发人员的消息区分等。可以通过配置版本号实现：

services.AddCap(x =>
{
   x.Version = "v1";
});

优化的雪花算法

雪花算法用于生成消息唯一标识，CAP使用的雪花算法是优化过的，解决了时钟漂移问题以及默认使用了Mac低10位作为WorkerId保证不会重复。同时也支持自定义雪花算法实现，如下：

services.AddSingleton<ISnowflakeId, SnowflakeId>();

消息自动清理

CAP 有自动清理机制来确保数据库表的数据量不会无限增长。默认情况下，对于成功处理的消息会在 1 天后过期然后清理，对于处理失败的消息会在 15 天后过期然后清理。你可以通过 SucceedMessageExpiredAfter 及 FailedMessageExpiredAfter 来自定义他们的过期时间。

清理器被 5 分钟扫描一次过期的消息，你可以通过 CollectorCleaningInterval 配置项来自定义间隔时间。

消费者特性

Attribute 订阅

基于 Attribute 的订阅可以减少重复代码，自动完成消息订阅和处理的绑定，这种方式可以让开发者专注于业务逻辑，消息传递的细节，提高代码可读性和维护性。

使用Attribute标记的方式进行消息订阅和消费：

[CapSubscribe("test.show.time")]
public void Show(DateTime datetime)
{
   Console.WriteLine(datetime);
}

多Attribute 订阅

CAP允许在同一方法上使用多个Attribute，可以让方法同时订阅多个不同的消息主题，从而增加了系统的灵活性和可扩展性及代码重用性。

[CapSubscribe("test.show.time")]
[CapSubscribe("test.show.hello")]
public void Show(DateTime datetime)
{
   Console.WriteLine(datetime);
}

通配符订阅

支持通过通配符进行消息订阅，尤其是在处理大量主题或复杂的消息路由需求时，可以大大简化订阅管理。

你可以使用 * 和 # 通配符来进行批量订阅。例如订阅所有以"test."开头的消息：

可以代替一个或多个词， # 可以代替零或多个词

[CapSubscribe("test.*")]
public void CheckReceivedMessage(DateTime datetime)
{
   Console.WriteLine(datetime);
}

异步订阅

支持异步方法订阅消息，可传递 CancellationToken 支持取消，这可以和启动停止Api配合实现灵活的并发处理及充分的资源利用。

[CapSubscribe("test.*")]
public async Task CheckReceivedMessage(DateTime datetime, CancellationToken cancellationToken)
{
   Console.WriteLine(datetime);
}

多程序集订阅

在单个系统进行模块化设计时，一般遵循分离关注点，订阅者可能分布在不同的程序集，CAP 支持跨多个程序集进行消息订阅，只需在不同程序集的类上定义订阅方法，然后使用AddSubscriberAssembly配置即可，然后CAP会自动查找发现注册。

services.AddCap(x =>
{
}).AddSubscriberAssembly(new Assembly[]{});

前缀订阅

订阅前缀用于设置在订阅时在父级类上指定一个前缀，那么类中的方法便自动继承此前缀，一般用于对消费者进行归类管理。

[CapSubscribe("customers")]
public class CustomersCapController : ICapSubscibe
{
	[CapSubscribe(".create", isPartial: true)]
	public Task HandleCreationAsync(Customer customer)	{
	}

	[CapSubscribe(".update", isPartial: true)]
	public Task HandleUpdateAsync(Customer customer)	{
	}
}

消费组

消费者组是由 Kafka 提出的一个非常重要的概念，用于实现高效的消息消费管理和负载均衡。CAP同样支持消费组的概念，而且针对了不同的消息队列进行适配，从而可以使用户获得一致性的使用体验。

[CapSubscribe("test.show.time", Group = "group1")]
public void CheckReceivedMessage(DateTime datetime)
{
   Console.WriteLine(datetime);
}

扇出(Fanout)消费

借助于抽象出的消费者组的概念，CAP支持扇出消费消息，也就是一发多收。

如果多个相同消费者组的消费者消费同一个Topic消息的时候，只会有一个消费者被执行。相反，如果消费者都位于不同的消费者组，则所有的消费者都会被执行。
下面的实例中，Hello1 和 Hello2 位于不同的 Group 中，那么都会收到消息，从而可以实现一发多收。

 [CapSubscribe("hello", Group = "bar")]
   public void Hello1(){
     Console.WriteLine("hello 1");
   }
   
 [CapSubscribe("hello", Group = "foo")]
   public void Hello2(){
     Console.WriteLine("hello 2");
   }

隐式类型消费

消费者参数接收支持隐式的消息类型转换，例如你发送的是一个 string 类型的日期字符串，那么在消费者端你也可以使用DateTime来进行接收。

await cap.PublishAsync<string>("hello", "2020/10/10 11:11");

[CapSubscribe("hello")]
public void Hello(DateTime time){
 Console.WriteLine($"hello {time}");
}

无类型消费

如果你不知道发布的消息类型或者不想再去定义类型，CAP支持使用 System.Text.Json.JsonElement 接收一切类型。

await _capBus.PublishAsync("hello", DateTime.Now);
await _capBus.PublishAsync("hello", 1345);
await _capBus.PublishAsync("hello", true);
await _capBus.PublishAsync("hello", "hello");
await _capBus.PublishAsync("hello", new Person{Name = "Jack", Age = 22});

[CapSubscribe("hello")]
public void Hello(JsonElement obj){
	Console.WriteLine(obj.ToString());
}

接口化类型消费

CAP支持通过扩展来实现基于接口的类型化消息消费，请查看 Sample.RabbitMQ.SqlServer 中的示例。

串行消费

大部分场景中希望发送和处理消息的顺序是一致的，在CAP中消息默认按顺序串行处理。

并行消费

在一些业务场景中消息顺序可能不是非常重要，但需要快速处理完消息，通过下面的配置项来开启并行处理。

支持启用多个订阅并行执行，通过 EnableSubscriberParallelExecute 来开启。

当开启后，支持通过 SubscriberParallelExecuteThreadCount 配置项设置线程数。

在内部CAP实现原理是先将消费线程收到的消息放置到缓冲区然后决定串行或者并行执行，因此还提供了另外一个配置参数 SubscriberParallelExecuteBufferFactor 来设置缓冲区的大小，这是一个乘数因子，缓冲区的大小值为（SubscriberParallelExecuteThreadCount * SubscriberParallelExecuteBufferFactor）。

背压机制

消息发送或消费的吞吐量除和用户的代码有关系外，还和当前的硬件负载及处理能力有关，缓冲区的设计在一定程度上能够最大化消息处理的吞吐能力及资源利用能力，但这并非万能的解药。当缓冲区满时CAP将自动启用背压机制来降低响应能力，这可以确保内存始终是安全的而不会引发OOM。

简单举例来说，有一个池子，一边向池子注水一边从池子放水，当注水的速度大于放水的速度时，那么池子满了后就会慢慢溢出。背压机制相当于给注水口添加了一个控制阀门，当池子满后将控制阀门流量来保持注水和放水平衡。

独立消费

有一些情况下某些消费者需要执行非常长的时间，这样会卡住另外一些执行时间比较短消费者，导致后续的消息不能及时消费。
所以在一些业务场景下需要某些 long-running 的消费者具有独立的数据管道以和其他正常的消费者隔离开，做到独立执行。

在 CAP 中可以在订阅时通过在 Attribute 添加 GroupConcurrent 参数来做到这一点，这是一个并行参数意味着也支持多个线程执行。

[CapSubscribe("hello", GroupConcurrent = 2)]
public void Hello2(){
  Console.WriteLine("hello 2");
  Thread.Sleep(1000);
}

在 Hello2中设置为 2 那么就代表这个方法在有新消息到达时候，最多会有2个线程在同时执行它

控制与集成特性

启动/停止 API

默认情况下 CAP 随 ASP.NET Core 宿主进程启动，如果你想手动控制启动或停止的时刻，你可以通过 IBootstrapper 接口来动态控制，在这一些需要特定时间来处理消息的场景中很方便。

提供启动和停止消息处理的API，便于控制消息处理的生命周期。

[Route("~/control/start")]
public async Task<IActionResult> Start([FromServices]IBootstrapper bootstrapper){
    await bootstrapper.BootstrapAsync();
    return Ok();
}

[Route("~/control/stop")]
public async Task<IActionResult> Stop([FromServices] IBootstrapper bootstrapper){
    await bootstrapper.DisposeAsync();
    return Ok();
}

异构系统集成

发送的消息可能来自于第三方异构系统而不是由CAP的发送端产生，CAP支持与异构系统进行消息对接。

如果异构系统发送的消息没有传递消费必须的Header信息，可以使用 CustomHeadersBuilder 配置项进行添加。

x.UseRabbitMQ(z => {
	z.CustomHeadersBuilder = (msg, sp) =>
	[
		new(Headers.MessageId, sp.GetRequiredService<ISnowflakeId>().NextId().ToString()),
		new(Headers.MessageName, msg.RoutingKey)
	];
});

多种传输支持

支持多种支持消息队列作为传输（Transport），以下是目前支持的传输。

RabbitMQ
Kafka
Azure Service Bus
Amazon SQS
NATS
In-Memory Queue
Redis Streams
Apache Pulsar

除此之外，下面是社区支持的传输，感谢贡献者。

ActiveMQ (@Lukas Zhang)： https://github.com/lukazh
RedisMQ (@木木)： https://github.com/difudotnet/CAP.RedisMQ.Extensions
ZeroMQ (@maikebing)： https://github.com/maikebing/CAP.Extensions
MQTT (@john jiang)： https://github.com/jinzaz/jinzaz.CAP.MQTT

多种存储支持

支持多种流行的消息存储方式，包括 SQL Server、MySQL、PostgreSQL、MongoDB、InMemory 等。

除了 InMemory 外，均提供了对事务消息的支持。

除此之外，下面是社区支持的传输，感谢贡献者。

SQLite (@colinin) ：https://github.com/colinin/DotNetCore.CAP.Sqlite
LiteDB (@maikebing) ：https://github.com/maikebing/CAP.Extensions
SQLite & Oracle (@cocosip) ：https://github.com/cocosip/CAP-Extensions
SmartSql (@xiangxiren) ：https://github.com/xiangxiren/SmartSql.CAP

监控，诊断，可观察性

CAP提供了 Dashboard 供用户快速的消息查看及实时消息发送及消费情况。
CAP提供了 Diagnostics API 供用户方便的使用 Logging、Tracing 及 Metrics.
CAP提供了 OpenTelemetry 可观察性标准的支持。

Dashboard 包含以下特性：

多语言
实时监控和可视化
手动查看与重试
与 ASP.NET Core紧密集成的授权
服务代理
服务发现之Consul
服务发现之Kubernetes
Metrics度量支持

多语言

提供中文和英文语言支持，在界面右上角进行切换显示。

实时监控和可视化

支持实时显示消息处理状态的图表，帮助监控消息生产及消费执行情况。

手动查看与重试

基于Dashboard提供的功能，可以查看消息的详细信息包括 header 和 content 内容。

在界面上可以对成功或失败的消息在手动再次进行发送或消费。

Subscriber 页面提供了的当前服务下所有订阅方法的查看，基于 Group 进行分组显示，便于用户进行统一查看于搜索。

与 ASP.NET Core 授权紧密集成

Dashboard 支持与 ASP.NET Core的授权机制无缝衔接。

匿名访问

cap.UseDashboard(d =>{
	d.AllowAnonymousExplicit = true;
});

OpenId Connect

services.AddAuthorization(options => { 
	options.AddPolicy(DashboardAuthorizationPolicy, policy => policy
		.AddAuthenticationSchemes(OpenIdConnectDefaults.AuthenticationScheme)
		.RequireAuthenticatedUser());
})
.AddAuthentication(opt => opt.DefaultScheme = CookieAuthenticationDefaults.AuthenticationScheme)
.AddCookie()
.AddOpenIdConnect(options => {
	...
});

services.AddCap(cap => {
	cap.UseDashboard(d => {
		d.AuthorizationPolicy = DashboardAuthorizationPolicy;
	});
});

请查看 Sample.Dashboard.Auth 示例项目了解更多。

Consul 节点发现

CAP是去中心化的设计，因此要查看其他节点的数据而不打开其他节点的Dashboard的情况下就要支持节点发现，Dashboard内置了代理功能用于代理其他节点API返回在数据在当前节点的Dashboard显示。
通过简单的节点切换即可来查看数据，从而可以带来更好的用户体验。

支持通过Consul进行节点发现，在过去大部分公司使用它来作为分布式系统的配置中心或注册中心。

更多信息，请查看官方文档。

K8s 节点发现

现代的应用程序基本是K8S作为默认集群管理及服务注册中心，DotNetCore.CAP.Dashboard.K8s 包提供了对 K8s 节点发现的支持。

CAP可读取K8S特定命名空间下的所有服务，然后 ping 出启用了Dashboard API 的那些节点转而代理切换过去。

独立运行的Dashboard

Dashboard 也支持不依赖服务项目而独立部署，此时无需再配置 AddCap 而直接使用下面的方式：

services.AddCapDashboardStandalone();

更多信息，请查看官方文档。

Diagnostics API

CAP 对 .NET DiagnosticSource 提供了支持，监听器名称为 CapDiagnosticListener。

你可以在 DotNetCore.CAP.Diagnostics.CapDiagnosticListenerNames 类下面找到CAP已经定义的事件名称。

Diagnostics 提供对外提供的事件信息有：

Before	After	Exception
消息持久化之前	消息持久化之后	消息持久化异常
消息向MQ发送之前	消息向MQ发送之后	消息向MQ发送异常
消息从MQ消费保存之前	消息从MQ消费保存之后
订阅者方法执行之前	订阅者方法执行之后	订阅者方法执行异常

CAP 对 .NET EventSource 提供了支持，计数器名称为 DotNetCore.CAP.EventCounter。

Metrics 提供了以下几个指标：

每秒发布速度
每秒消费速度
每秒调用订阅者速度
每秒执行订阅者平均耗时

OpenTelemetry 支持

分布式追踪在现代应用中有着非常重要的作用，OpenTelemetry 是一个开放的标准支持多种编程语言和框架，提供了跨平台、跨语言的兼容性，便于在异构系统中实现统一的监控和追踪。

CAP可以衔接上游触发的消息源链路，然后经由每个服务，每个传输，每个存储后传递到下游执行，整个过程都会被记录并还原，形成完整的调用链，便于性能优化和问题排查。除此之外，在消息消费时再次发送的消息链路同样可以被准确衔接，做到调用链的完整准确。

CAP 通过 DotNetCore.CAP.OpenTelemetry 包提供了对 OpenTelemetry 的支持。

services.AddOpenTelemetryTracing((builder) => builder
    .AddAspNetCoreInstrumentation()
    .AddCapInstrumentation()    
    .AddZipkinExporter()
);