.NET CAP入门到入土

CAP

什么是CAP？

CAP 是一个EventBus ，同时也是一个再微服务或者SOA系统中解决分布式事务问题的一个框架。它有助于创建可扩展，可靠而且易于更改的微服务系统。

什么是EventBus？
事件总线是一种机制，它允许不同的组件彼此通信而不彼此了解。组件可以将事件发送到Eventbus,而无需知道是谁来接听或有多少其他人来接听。组件可以侦听Eventbus 上的事件。这样，组件可以相互通信而无需相互依赖。同样，很容易替换一个组件。只要新的组件理解正在发送和接收的事件，其他组件就永远不会知道。

相对于其他的Service Bus 或者 Event Bus ,CAP 拥有自己的特色 ，它不要求使用者发送消息或者处理消息的时候实现或者继承任何接口，拥有非常高的灵活性。它一直以约定大于配置的方式存在，所以CAP 使用起来非常简单，并且拥有轻量级。

CAP 采用模块化设计，具有高度的可扩展性。这里可以有许多选项可以选择，包括消息队列、存储、序列化方式等。系统的许多元素内从可以替换为自定义实现。

快速开始

1. 安装

PM>Install-PackageDotNetCore.CAP

2. 在Asp.Net Core中集成

这里以便于快速启动，我们使用内存事件存储和消息队列

PM>Install-PackageDotNetCore.CAP.InMemoryStorage PM>Install-PackageSavorboard.CAP.InMemoryMessageQueue

3. 在Startup.cs中，添加以下配置

public void ConfigureServices(IServiceCollection services) { services.AddCap(x => { x.UseInMemoryStorage(); x.UseInMemoryMessageQueue(); }); }

4. 发送消息

public class PublishController : Controller { [Route("~/send")] public IActionResult SendMessage([FromServices]ICapPublisher capBus) { capBus.Publish("test.show.time", DateTime.Now); return Ok(); } }

5. 发送延迟消息

public class PublishController : Controller { [Route("~/send/delay")] public IActionResult SendDelayMessage([FromServices]ICapPublisher capBus) { capBus.PublishDelay(TimeSpan.FromSeconds(100),"test.show.time", DateTime.Now); return Ok(); } }

6. 发送带有头信息的消息

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

7. 处理消息

public class ConsumerController : Controller { [NonAction] [CapSubscribe("test.show.time")] public void ReceiveMessage(DateTime time) { Console.WriteLine("message time is:" + time); } }

8. 处理包含头消息的消息

[CapSubscribe("test.show.time")] public void ReceiveMessage(DateTime time, [FromCap]CapHeader header) { Console.WriteLine("message time is:" + time); Console.WriteLine("message firset header :" + header["my.header.first"]); Console.WriteLine("message second header :" + header["my.header.second"]); }

配置

默认情况下，你在向DI容器中注册CAP服务的时候指定此配置。

services.AddCap(config=> { // config.XXX });

其中services代表的是IServiceCollection接口对象，它位于Microsoft.Extensions.DependencyInjection下面。

什么是最低配置？

最简单的回答就是，至少你要配置一个传输器和一个存储，如果你想快速开始你可以使用下面的配置：

services.AddCap(config => { config.UseInMemoryMessageQueue(); //需要引用 Savorboard.CAP.InMemoryMessageQueue 包 config.UseInMemoryStorage(); });

有关具体的传输器配置和存储配置，你可以查看 Transports 章节和 Persistent 章节中具体组件提供的配置项。

订阅者中的配置

订阅者使用[CapSubscribe]这个Attribute来标记成为一个订阅者，订阅者可以位于 ASP.NET Core 的 Controller 或 Service 中。

当你在声明[CapSubscribe]时候，可以通过指定以下参数来改变订阅者的行为。

Name

string, 必须项

通过指定Name参数来订阅消息，其对应发布消息时通过 Publish(“Name”) 指定的名称。

该名称在不同的 Broker 有不同的对应项。

• 在 RabbitMQ 中对应 Routing Key。
• 在 Kafka 中对应 Topic。
• 在 AzureServiceBus 中对应 Subject。
• 在 NATS 中对应 Subject。
• 在 RedisStrems 中对应 Stream.

Group

string, 可选项

通过指定Group参数来使订阅者位于单独的消费者组中，消费者组的概念类似于 Kafka 中的消费者组。如果不指定此参数将使用当前程序集名称(DefaultGroupName)作为默认值。

相同Name的订阅者设置为不同的组时，他们都会收到消息。相反如果相同Name的订阅者设置相同的组时，只有一个会收到消息。

不同Name的订阅者设置为不同的组时，也是有意义的，他们可以拥有独立的线程来执行。相反如果不同Name的订阅者设置相同的组时，他们将共享消费线程。

Group 在不同的 Broker 有不同的对应项。

• 在 RabbitMQ 中对应 Queue。
• 在 Kafka 中对应 Consumer Group。
• 在 AzureServiceBus 中对应 Subscription Name。
• 在 NATS 中对应 Queue Group。
• 在 RedisStrems 中对应 Consuemr Group.

GroupConcurrent

byte, 可选项

通过指定GroupConcurrent参数的值来设置订阅者并行执行的并行度。并行执行意味着其需要位于独立线程中，因此如果你没有指定Group参数，则 CAP 将会以Name的值自动创建一个 Group。

如果你有多个订阅者都设置为了相同的 Group，并且也给订阅者都设置了GroupConcurrent的值，则并行度为组内值的和。
本设置只对新消息生效，重试的消息不受并行度限制。

自定义配置项

在AddCap中CapOptions对象是用来存储配置相关信息，默认情况下它们都具有一些默认值，有些时候你可能需要自定义。

DefaultGroupName

默认值：cap.queue.{程序集名称}

默认的消费者组的名字，在不同的 Transports 中对应不同的名字，可以通过自定义此值来自定义不同 Transports 中的名字，以便于查看。

GroupNamePrefix

默认值：Null

为订阅 Group 统一添加前缀。

TopicNamePrefix

默认值： Null

为 Topic 统一添加前缀。

Version

默认值：v1

用于给消息指定版本来隔离不同版本服务的消息，常用于A/B测试或者多服务版本的场景。以下是其应用场景：

FailedRetryInterval

默认值：60 秒

在消息发送的时候，如果发送失败，CAP将会对消息进行重试，此配置项用来配置每次重试的间隔时间。

在消息消费的过程中，如果消费失败，CAP将会对消息进行重试消费，此配置项用来配置每次重试的间隔时间。

UseStorageLock

默认值: false

如果设置为true，我们将使用基于数据库的分布式锁以应对重试进程在多个实例下对数据库数据的并发获取问题。这将会在数据库生成 cap.lock 表。该锁只保护重试消息的获取过程；它不保证每条失败消息在整个集群中最多每FailedRetryInterval重试一次。

ConsumerThreadCount

默认值：1

消费者线程并行处理消息的线程数，当这个值大于1时，将不能保证消息执行的顺序。

CollectorCleaningInterval

默认值：300 秒

收集器删除已经过期消息的时间间隔。

SchedulerBatchSize

默认值：1000

调度器每次循环获取的延迟或排队消息的最大数量。

FailedRetryCount

默认值：50

重试的最大次数。当达到此设置值时，将不会再继续重试，通过改变此参数来设置重试的最大次数。

FallbackWindowLookbackSeconds

默认值：240 秒

配置重试处理器拾取Scheduled或Failed状态消息的回退时间窗。

FailedThresholdCallback

默认值：NULL

类型：Action<FailedInfo>

重试阈值的失败回调。当重试达到 FailedRetryCount 设置的值的时候，将调用此 Action 回调，你可以通过指定此回调来接收失败达到最大的通知，以做出人工介入。例如发送邮件或者短信。

SucceedMessageExpiredAfter

默认值：24*3600 秒（1天后）

成功消息的过期时间（秒）。 当消息发送或者消费成功时候，在时间达到SucceedMessageExpiredAfter秒时候将会从 Persistent 中删除，你可以通过指定此值来设置过期的时间。

FailedMessageExpiredAfter

默认值：15243600 秒（15天后）

失败消息的过期时间（秒）。 当消息发送或者消费失败时候，在时间达到FailedMessageExpiredAfter秒时候将会从 Persistent 中删除，你可以通过指定此值来设置过期的时间。

[已移除] UseDispatchingPerGroup

默认值: false

版本 8.2.0 中移除，已是默认行为。

默认情况下，CAP会将所有消费者组的消息都先放置到内存同一个Channel中，然后线性处理。 如果设置为 true，则每个消费者组都会根据ConsumerThreadCount设置的值创建单独的线程进行处理。

[已过时] EnableConsumerPrefetch

默认值: false， 在 7.0 版本之前默认行为 true

该配置项已被重命名为EnableSubscriberParallelExecute，请使用新选项。

EnableSubscriberParallelExecute

默认值: false

如果设置为true，CAP将提前从Broker拉取一批消息置于内存缓冲区，然后执行订阅方法；当订阅方法执行完成后，拉取下一批消息至于缓冲区然后执行。

SubscriberParallelExecuteThreadCount

Default:Environment.ProcessorCount

当启用EnableSubscriberParallelExecute时, 可通过此参数执行并行处理的线程数，默认值为处理器个数。

SubscriberParallelExecuteBufferFactor

Default: 1

当启用EnableSubscriberParallelExecute时, 通过此参数设置缓冲区和线程数的因子系数，也就是缓冲区大小等于SubscriberParallelExecuteThreadCount乘SubscriberParallelExecuteBufferFactor.

EnablePublishParallelSend

默认值: false

默认情况下，发送的消息都先放置到内存同一个Channel中，然后线性处理。 如果设置为 true，则发送消息的任务将由.NET线程池并行处理，这会大大提高发送的速度。

消息

发送&处理消息这里可以参考上面的快速开始来完成

补偿事务

Compensating transaction

某些情况下，消费者需要返回值以告诉发布者执行结果，以便于发布者实施一些动作，通常情况下这属于补偿范围。

你可以在消费者执行的代码中通过重新发布一个新消息来通知上游，CAP 提供了一种简单的方式来做到这一点。 你可以在发送的时候指定callbackName来得到消费者的执行结果，通常这仅适用于点对点的消费。以下是一个示例。

例如，在一个电商程序中，订单初始状态为 pending，当商品数量成功扣除时将状态标记为 succeeded ，否则为 failed。

// ============= Publisher ================= _capBus.Publish("place.order.qty.deducted", contentObj: new { OrderId = 1234, ProductId = 23255, Qty = 1 }, callbackName: "place.order.mark.status"); // publisher using `callbackName` to subscribe consumer result [CapSubscribe("place.order.mark.status")] public void MarkOrderStatus(JsonElement param) { var orderId = param.GetProperty("OrderId").GetInt32(); var isSuccess = param.GetProperty("IsSuccess").GetBoolean(); if(isSuccess){ // mark order status to succeeded } else{ // mark order status to failed } } // ============= Consumer =================== [CapSubscribe("place.order.qty.deducted")] public object DeductProductQty(JsonElement param) { var orderId = param.GetProperty("OrderId").GetInt32(); var productId = param.GetProperty("ProductId").GetInt32(); var qty = param.GetProperty("Qty").GetInt32(); //business logic return new { OrderId = orderId, IsSuccess = true }; }

控制回调响应

你可以通过[FromCap]标记在订阅方法中注入CapHeader参数，并利用其提供的方法来向回调上下文中添加额外的头信息或者终止回调。

示例如下：

[CapSubscribe("place.order.qty.deducted")] public object DeductProductQty(JsonElement param, [FromCap] CapHeader header) { var orderId = param.GetProperty("OrderId").GetInt32(); var productId = param.GetProperty("ProductId").GetInt32(); var qty = param.GetProperty("Qty").GetInt32(); // 添加额外的头信息到响应消息中 header.AddResponseHeader("some-message-info", "this is the test"); // 或再次添加回调的回调 header.AddResponseHeader(DotNetCore.CAP.Messages.Headers.CallbackName, "place.order.qty.deducted-callback"); // 如果你不再遵从发送着指定的回调，想修改回调，可通过 RewriteCallback 方法修改。 header.RewriteCallback("new-callback-name"); // 如果你想终止/停止，或不再给发送方响应，调用 RemoveCallback 来移除回调。 header.RemoveCallback(); return new { OrderId = orderId, IsSuccess = true }; }

异构系统集成

在 3.0+ 版本中，我们对消息结构进行了重构，我们利用了消息队列中消息协议中的 Header 来传输一些额外信息，以便于在 Body 中我们可以做到不需要修改或包装使用者的原始消息数据格式和内容进行发送。

这样的做法是合理的，它有助于在异构系统中进行更好的集成，相对于以前的版本使用者不需要知道CAP内部使用的消息结构就可以完成集成工作。

现在我们将消息划分为 Header 和 Body 来进行传输。

Body 中的数据为用户发送的原始消息内容，也就是调用 Publish 方法发送的内容，我们不进行任何包装仅仅是序列化后传递到消息队列。

在 Header 中，我们需要传递一些额外信息以便于CAP在收到消息时能够提取到关键特征进行操作。

以下是在异构系统中，需要在发消息的时候向消息的Header 中写入的内容：

以 Java 系统发送 RabbitMQ 为例：

ap<String,Object>headers=newHashMap<String,Object>();headers.put("cap-msg-id",UUID.randomUUID().toString());headers.put("cap-msg-name",routingKey);channel.basicPublish(exchangeName,routingKey,newAMQP.BasicProperties.Builder().headers(headers).build(),messageBodyBytes);// messageBodyBytes = "发送的json".getBytes(Charset.forName("UTF-8"))// 注意 messageBody 默认为 json 的 byte[]，如果采用其他系列化，需要在CAP侧自定义反序列化器

消息重试

重试在整个CAP架构设计中具有重要作用，CAP 中会针对发送失败或者执行失败的消息进行重试。在整个 CAP 的设计过程中有以下几处采用的重试策略。

1、 发送重试

在消息发送过程中，当出现 Broker 宕机或者连接失败的情况亦或者出现异常的情况下，这个时候 CAP 会对发送的重试，第一次重试次数为 3，4分钟后以后每分钟重试一次，进行次数 +1，当总次数达到50次后，CAP将不对其进行重试。

2、 消费重试

当 Consumer 接收到消息时，会执行消费者方法，在执行消费者方法出现异常时，会进行重试。这个重试策略和上面的 发送重试 是相同的。

消息数据清理

据库消息表中具有一个 ExpiresAt 字段表示消息的过期时间，当消息发送成功或者消费成功后，CAP 会将消息状态为 Successed 的 ExpiresAt 设置为 1天 后过期，会将消息状态为 Failed 的 ExpiresAt 设置为 15天 后过期（可通过 FailedMessageExpiredAfter 配置)。

CAP 默认情况下会每隔5分钟将消息表的数据进行清理删除，避免数据量过多导致性能的降低。清理规则为 ExpiresAt 不为空并且小于当前时间的数据。 也就是说状态为Failed的消息（正常情况他们已经被重试了 50 次），如果你15天没有人工介入处理，同样会被清理掉。你可以通过 CollectorCleaningInterval 配置项来自定义间隔时间。

过滤器

自定义过滤器

添加过滤器

创建一个过滤器类，并继承SubscribeFilter抽象类。

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

在一些场景中，如果想终止订阅者方法执行，可以在OnSubscribeExecutingAsync中抛出异常，并且在OnSubscribeExceptionAsync中选择忽略该异常。

通过在ExceptionContext中设置context.ExceptionHandled = true来忽略异常。

public override Task OnSubscribeExceptionAsync(ExceptionContext context) { context.ExceptionHandled = true; }

配置过滤器

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

目前不支持同时添加多个过滤器。

序列化

CAP 提供了ISerializer接口来支持对消息进行序列化，默认情况下我们使用 json 来对消息进行序列化处理并存储到数据库中。

自定义序列化

public class YourSerializer: ISerializer { Task<TransportMessage> SerializeAsync(Message message) { } Task<Message> DeserializeAsync(TransportMessage transportMessage, Type valueType) { } }

然后将你的实现注册到容器中:

/注册你的自定义实现 services.AddSingleton<ISerializer, YourSerializer>(); // --- services.AddCap

事务

异步确保

异步确保这种方案又叫做本地消息表，这是一种经典的方案，方案最初来源于 eBay，参考资料见段末链接。这种方案目前也是企业中使用最多的方案之一。

传输

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ 服务器是用 Erlang 语言编写的，而聚类和故障转移是构建在开源的通讯平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

CAP 支持使用 RabbitMQ 作为消息传输器。

配置

要使用 RabbitMQ 作为消息传输器，你需要从 NuGet 安装以下扩展包：

Install-PackageDotNetCore.CAP.RabbitMQ

然后，你可以在Startup.cs的ConfigureServices方法中添加基于 RabbitMQ 的配置项。

public void ConfigureServices(IServiceCollection services) { // ... services.AddCap(x => { x.UseRabbitMQ(opt=> { //RabbitMQOptions }); // x.UseXXX ... }); }

RabbitMQ Options

CAP 直接对外提供的 RabbitMQ 配置参数如下：

ConnectionFactory Option¶

如果你需要更多原生ConnectionFactory相关的配置项，可以通过ConnectionFactoryOptions配置项进行设定：

services.AddCap(x => { x.UseRabbitMQ(o => { o.HostName = "localhost"; o.ConnectionFactoryOptions = opt => { //rabbitmq client ConnectionFactory config }; }); });

CustomHeadersBuilder Option

当需要从异构系统或者直接接收从RabbitMQ 控制台发送的消息时，由于 CAP 需要定义额外的头信息才能正常订阅，所以此时会出现异常。通过提供此参数来进行自定义头信息的设置来使订阅者正常工作。

用法如下：

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

如何连接 RabbitMQ 集群？

使用逗号分隔连接字符串即可，如下：

x=> x.UseRabbitMQ("localhost:5672,localhost:5673,localhost:5674")

存储

MySql

MySQL 是一个开源的关系型数据库，你可以使用 MySQL 来作为 CAP 消息的持久化。

配置

要使用 MySQL 存储，你需要从 NuGet 安装以下扩展包：

Install-PackageDotNetCore.CAP.MySql

然后，你可以在Startup.cs的ConfigureServices方法中添加基于内存的配置项。

public void ConfigureServices(IServiceCollection services) { // ... services.AddCap(x => { x.UseMySql(opt=>{ //MySqlOptions }); // x.UseXXX ... }); }

配置项

自定义表名

你可以通过重写IStorageInitializer接口获取表名称的方法来做到这一点

示例代码：

{ public override string GetPublishedTableName() { //你的 发送消息表 名称 } public override string GetReceivedTableName() { //你的 接收消息表 名称 } }

然后将你的实现注册到容器中

services.AddSingleton<IStorageInitializer, MyTableInitializer>();

使用事务发布消息

ADO.NET

private readonly ICapPublisher _capBus; using (var connection = new MySqlConnection(AppDbContext.ConnectionString)) { using (var transaction = connection.BeginTransaction(_capBus, autoCommit: false)) { //your business code connection.Execute("insert into test(name) values('test')", transaction: (IDbTransaction)transaction.DbTransaction); _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now); transaction.Commit(); } }

EntityFramework

private readonly ICapPublisher _capBus; using (var trans = dbContext.Database.BeginTransaction(_capBus, autoCommit: false)) { dbContext.Persons.Add(new Person() { Name = "ef.transaction" }); _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now); dbContext.SaveChanges(); trans.Commit(); }