案排渭牟可能有些人还不知道 CAP 是什么老规矩来一个简介。CAP 是一个用来解决微服务或者分布式系统中分布式事务问题的一个开源项目解决方案https://github.com/dotnetcore/CAP同样可以用来作为 EventBus 使用该项目诞生于2016年目前在 Github 已经有超过 6500 Star 和 110 贡献者以及在 NuGet超 800 万的下载量并在越来越多公司的和项目中得到应用。如果你想对 CAP 更多了解请查看我们的 官方文档。本次在 CAP 8.2 版本中我们主要带来了以下新特性支持消费者独立并行执行CapHeader 添加更多行为来控制回调改进当启用 EnablePublishParallelSend 时默认的调度行为为 NATS 支持 CustomHeadersBuilder 配置项其他改进支持消费者独立并行执行在介绍这个新特性之前 我们先来看看过去如果想让消费者独立执行应该怎么做。熟悉 CAP 的同学可能了解到CAP提供了一个 UseDispatchingPerGroup 配置项用来为消费者组启用独立的调度器独立的调度器意味着其具有独立的消费管道这样消费者组之间就不会互相影响默认如果不开启此配置项会将不同组的消息放置到同一管道来调度并串行执行从而可以实现如某些消费者执行非常长的时间也不会卡住另外一些执行时间比较短消费者这样也就做到了独立执行。当然如果想开启多个线程并发执行可以通过设置 SubscriberParallelExecuteThreadCount 配置项来启用多个执行线程执行这会提高消息的消费速度。可以看到在过去是不支持为消费者设置独立的并行度的SubscriberParallelExecuteThreadCount 其提供的是全局配置不支持独立为订阅者进行设置。以上都是过去的行为下面我们看一下现在如何做。在新版本中我们为 [CapSubscribe] 这个 Attribute 添加了一个新的参数 GroupConcurrent 来为订阅着设置并行度也就是说每一个组/订阅者都可以独立设置设置之后他们之间互不影响独立执行。下面我们来看一个示例定义消费者如下[CapSubscribe(hello)]public void Hello1(){Console.WriteLine(hello 1);}[CapSubscribe(hello, Group foo, GroupConcurrent 2)]public void Hello2(){Console.WriteLine(hello 2);Thread.Sleep(1000);}[CapSubscribe(hello, GroupConcurrent 2)]public void Hello3(){Console.WriteLine(hello 3);Thread.Sleep(1000);}为了验证 Hello2 和 Hello3 的执行我添加了 Thread.Sleep(1000);这2行代码。现在发送一条名为hello的消息我们来看一下订阅者是如何执行的。await cap.PublishAsync(hello, null);熟悉CAP的同学可能知道在过去 Hello1 和 Hello2 是都会执行如果启用了 UseDispatchingPerGroup 配置项那么会同时执行否则会串行执行。在新版本我们来看看是怎么执行的。先看 Hello2 新添加了 GroupConcurrent 参数会是什么效果呢 正如和前面说过的这是一个消费并行度参数在 Hello2中设置为 2 那么就代表这个方法在有新消息到达时候最多会有2个线程在同时执行它。所以此处的打印结果会是同时打印2个 hello 2。再来看 Hello1 和 Hello3 这两个哪个会打印呢 其实这两个也都会打印出来那么他们两个有什么区别呢可以看到的是一个添加了 GroupConcurrent 另外一个没添加。所以我来解释一下在内部我们会检测如果一个订阅者添加了 GroupConcurrent 而没有指定 Group 的话那么我们会认为其需要独立执行此时我们会自动采用订阅名称为其生成默认 Group 名称在此示例中就相当于设置了 Grouphello。这就说完了整体来说是不是很简单需要就设置不需要就不设置没有复杂的配置。所以新版本我们移除了 UseDispatchingPerGroup 配置项如你所见这已经是默认行为。PS: 如果将多个订阅者设置为同一组并且还为这些订阅者设置了 GroupConcurrent 值则并行度为组内值的和。PS: GroupConcurrent 在 RabbitMQ 中会自动对应设置Qos无需再手动干预。CapHeader 添加更多行为来控制补偿事务在发布消息的时候我们支持指定 callbackName 参数来进行补偿事务消费者执行完成后 CAP 会自动调用 callbackName 来回调执行补偿方法。接收到用户的反馈在某些场景下消费者可能执行成功就不需要再回调发布方那么此时消费者就需要控制补偿行为来避免回调再或者想回调另外一个方法来执行其他的补偿流程。在过去可能需要通过消费过滤器来实现现在我们为 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 };}改进启用 EnablePublishParallelSend 时默认调度行为在发布消息时我们支持使用 EnablePublishParallelSend 来启用并行发送因为发布消息属于 short-term 的task所以非常放置到 .NET 线程池执行。这本身没有问题在过去我们进行过压力测试也没有发现问题但是零星用户反馈会导致应用崩溃询问也没有反馈更多信息PS: 大家提交完 issue 不是就代表不用管了还请关注后续暂时还得不到更多消息是否是 Docker 限制了内存或CPU导致的不得而知。无论如何我们在这个版本优化了这一行为现在会分批次来将任务提交到线程池在上一批次执行完成后才会放置下一批次的任务。如果发布速度很快内存队列达到阈值也就是堆满后会启用背压机制来延缓发布者的响应。为 NATS 支持 CustomHeadersBuilder 配置项现在 NATS 也有了 CustomHeadersBuilder 配置项用于和三方异构系统做对接。没有太多好说的和其他 Transport 的用法都是一致。其他改进升级 Npgsql 到最新版本安全性升级Npgsql 易受通过协议信息大小溢出进行 SQL 注入。https://github.com/npgsql/npgsql/security/advisories/GHSA-x9vc-6hfv-hg8c升级 Dashboard npm包到最新次要版本修复 NATS重启后健康检查偶尔未正确恢复的问题参考 Issue #1542总结以上就是本版本我们做出的一些新特性和改动感谢大家的支持我们很开心能够帮助到大家 。大家在使用的过程中遇到问题希望也能够积极的反馈帮助CAP变得越来越好。??如果你喜欢这个项目可以通过下面的连接点击 Star 给我们支持。GitHub stars