.NET CAP入门到入土

CAP什么是CAPCAP 是一个EventBus 同时也是一个再微服务或者SOA系统中解决分布式事务问题的一个框架。它有助于创建可扩展可靠而且易于更改的微服务系统。什么是EventBus事件总线是一种机制它允许不同的组件彼此通信而不彼此了解。组件可以将事件发送到Eventbus,而无需知道是谁来接听或有多少其他人来接听。组件可以侦听Eventbus 上的事件。这样组件可以相互通信而无需相互依赖。同样很容易替换一个组件。只要新的组件理解正在发送和接收的事件其他组件就永远不会知道。相对于其他的Service Bus 或者 Event Bus ,CAP 拥有自己的特色 它不要求使用者发送消息或者处理消息的时候实现或者继承任何接口拥有非常高的灵活性。它一直以约定大于配置的方式存在所以CAP 使用起来非常简单并且拥有轻量级。CAP 采用模块化设计具有高度的可扩展性。这里可以有许多选项可以选择包括消息队列、存储、序列化方式等。系统的许多元素内从可以替换为自定义实现。快速开始1. 安装PMInstall-PackageDotNetCore.CAP2. 在Asp.Net Core中集成这里以便于快速启动我们使用内存事件存储和消息队列PMInstall-PackageDotNetCore.CAP.InMemoryStorage PMInstall-PackageSavorboard.CAP.InMemoryMessageQueue3. 在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 Dictionarystring, 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]时候可以通过指定以下参数来改变订阅者的行为。Namestring, 必须项通过指定Name参数来订阅消息其对应发布消息时通过 Publish(“Name”) 指定的名称。该名称在不同的 Broker 有不同的对应项。在 RabbitMQ 中对应 Routing Key。在 Kafka 中对应 Topic。在 AzureServiceBus 中对应 Subject。在 NATS 中对应 Subject。在 RedisStrems 中对应 Stream.Groupstring, 可选项通过指定Group参数来使订阅者位于单独的消费者组中消费者组的概念类似于 Kafka 中的消费者组。如果不指定此参数将使用当前程序集名称(DefaultGroupName)作为默认值。相同Name的订阅者设置为不同的组时他们都会收到消息。相反如果相同Name的订阅者设置相同的组时只有一个会收到消息。不同Name的订阅者设置为不同的组时也是有意义的他们可以拥有独立的线程来执行。相反如果不同Name的订阅者设置相同的组时他们将共享消费线程。Group 在不同的 Broker 有不同的对应项。在 RabbitMQ 中对应 Queue。在 Kafka 中对应 Consumer Group。在 AzureServiceBus 中对应 Subscription Name。在 NATS 中对应 Queue Group。在 RedisStrems 中对应 Consuemr Group.GroupConcurrentbyte, 可选项通过指定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类型ActionFailedInfo重试阈值的失败回调。当重试达到 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如果设置为trueCAP将提前从Broker拉取一批消息置于内存缓冲区然后执行订阅方法当订阅方法执行完成后拉取下一批消息至于缓冲区然后执行。SubscriberParallelExecuteThreadCountDefault:Environment.ProcessorCount当启用EnableSubscriberParallelExecute时, 可通过此参数执行并行处理的线程数默认值为处理器个数。SubscriberParallelExecuteBufferFactorDefault: 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 中写入的内容键类型说明cap-msg-idlong消息Id 由雪花算法生成cap-msg-namestring消息名称即 Topic 名字cap-msg-typestring消息的类型, 即 typeof(T).FullName (非必须)cap-senttimestring发送的时间 (非必须)以 Java 系统发送 RabbitMQ 为例apString,ObjectheadersnewHashMapString,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 会对发送的重试第一次重试次数为 34分钟后以后每分钟重试一次进行次数 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 { // *** }.AddSubscribeFilterMyCapFilter();目前不支持同时添加多个过滤器。序列化CAP 提供了ISerializer接口来支持对消息进行序列化默认情况下我们使用 json 来对消息进行序列化处理并存储到数据库中。自定义序列化public class YourSerializer: ISerializer { TaskTransportMessage SerializeAsync(Message message) { } TaskMessage DeserializeAsync(TransportMessage transportMessage, Type valueType) { } }然后将你的实现注册到容器中:/注册你的自定义实现 services.AddSingletonISerializer, 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 OptionsCAP 直接对外提供的 RabbitMQ 配置参数如下配置项描述类型默认值HostName宿主地址如果要配置集群可以使用逗号分隔例如192.168.1.111,192.168.1.112stringlocalhostUserName用户名stringguestPassword密码stringguestVirtualHost虚拟主机string/Port端口号int-1ExchangeNameCAP默认Exchange名称stringcap.default.topicQueueArguments队列额外参数 x-argumentsQueueArgumentsOptionsN/AQueueOptions更改已创建队列的选项QueueRabbitOptions{ Durabletrue, Exclusivefalse, AutoDeletefalse }ConnectionFactoryOptionsRabbitMQClient原生参数ConnectionFactoryN/ACustomHeadersBuilder订阅者自定义头信息见下文N/APublishConfirms是否启用发布确认boolfalseBasicQosOptions指定消费的QosBasicQosN/AConnectionFactory 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.GetRequiredServiceISnowflakeId().NextId().ToString()), new(DotNetCore.CAP.Messages.Headers.MessageName, msg.RoutingKey) ]; });如何连接 RabbitMQ 集群使用逗号分隔连接字符串即可如下x x.UseRabbitMQ(localhost:5672,localhost:5673,localhost:5674)存储MySqlMySQL 是一个开源的关系型数据库你可以使用 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 ... }); }配置项NAMEDESCRIPTIONTYPEDEFAULTTableNamePrefixCap表名前缀stringcapConnectionString数据库连接字符串stringnull自定义表名你可以通过重写IStorageInitializer接口获取表名称的方法来做到这一点示例代码{ public override string GetPublishedTableName() { //你的 发送消息表 名称 } public override string GetReceivedTableName() { //你的 接收消息表 名称 } }然后将你的实现注册到容器中services.AddSingletonIStorageInitializer, MyTableInitializer();使用事务发布消息ADO.NETprivate 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(); } }EntityFrameworkprivate 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(); }