CAP 是基于 .NET Standard 的 C# 库，提供分布式事务解决方案及 EventBus 功能。支持 SQL Server、MySQL、PostgreSQL、MongoDB 等存储，兼容 RabbitMQ、Kafka、Azure Service Bus 等多种消息队列传输器。通过配置 EntityFramework 或 MongoDB 上下文及消息中间件连接信息，可在 Controller 或 Service 层利用 ICapPublisher 发布消息，并通过 CapSubscribeAttribute 或 ICapSubscribe 接口订阅处理消息，确保微服务系统中的数据最终一致性。
在分布式系统中，由于各个系统服务之间的独立性和网络通信的不确定性，要确保跨系统的事务操作的最终一致性是一项重大的挑战。本文介绍一个基于 .NET Standard 的 C# 库：CAP。
CAP 是一个基于 .NET Standard 的 C# 库，它是一种处理分布式事务的解决方案，同时具有 EventBus 的功能。它具有轻量级、易使用、高性能等特点。CAP 是一个 EventBus，同时也是在微服务或者 SOA 系统中解决分布式事务问题的一个框架。它有助于创建可扩展、可靠并且易于更改的微服务系统。
事件总线是一种机制，它允许不同的组件彼此通信而不彼此了解。组件可以将事件发送到 Eventbus，而无需知道是谁来接听或有多少其他人来接听。组件也可以侦听 Eventbus 上的事件，而无需知道谁发送了事件。这样，组件可以相互通信而无需相互依赖。同样，很容易替换一个组件。只要新组件了解正在发送和接收的事件，其他组件就永远不会知道。
SQL Server、MySQL、PostgreSQL、MongoDB、In-Memory Storage。
RabbitMQ、Kafka、Azure Service Bus、Amazon SQS、NATS、In-Memory Queue、Redis Streams、Apache Pulsar。
我本地安装的是 DotNetCore.CAP.RabbitMQ。
//你可以按需选择下面的包进行安装:
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
我本地安装的是 DotNetCore.CAP.MongoDB。
//按需选择安装你正在使用的数据库:
PM> Install-Package DotNetCore.CAP.SqlServer
PM> Install-Package DotNetCore.CAP.MySql
PM> Install-Package DotNetCore.CAP.PostgreSql
PM> Install-Package DotNetCore.CAP.MongoDB
builder.Services.AddCap(x =>
{
//如果你使用的 EF 进行数据操作，你需要添加如下配置：
//配置数据库上下文
x.UseEntityFramework<AppDbContext>();
//如果你使用的 MongoDB，你可以添加如下配置：
x.UseMongoDB("ConnectionStrings"); //注意，仅支持 MongoDB 4.0+ 集群
//CAP RabbitMQ 配置
x.UseRabbitMQ(rab => {
rab.HostName = "192.0.1.1";
rab.Password = "123456";
rab.Port = 5672;
rab.UserName = "123456";
});
});
在 Controller 中注入 ICapPublisher 然后使用 ICapPublisher 进行消息发送。
public class PublishController : Controller
{
private readonly ICapPublisher _capBus;
public PublishController(ICapPublisher capPublisher)
{
_capBus = capPublisher;
}
//不使用事务
[Route("~/without/transaction")]
public IActionResult WithoutTransaction()
{
_capBus.Publish("xxx.services.show.time", DateTime.Now);
// Publish delay message
_capBus.PublishDelayAsync(TimeSpan.FromSeconds(delaySeconds), "xxx.services.show.time", DateTime.Now);
return Ok();
}
//Ado.Net 中使用事务，自动提交
[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);
}
}
return Ok();
}
//EntityFramework 中使用事务，自动提交
[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();
}
}
在 Action 上添加 CapSubscribeAttribute 来订阅相关消息。
public class PublishController : Controller
{
[CapSubscribe("xxx.services.show.time")]
public void CheckReceivedMessage(DateTime datetime)
{
Console.WriteLine(datetime);
}
}
如果你的订阅方法没有位于 Controller 中，则你订阅的类需要继承 ICapSubscribe：
namespace xxx.Service
{
public interface ISubscriberService
{
void CheckReceivedMessage(DateTime datetime);
}
public class SubscriberService:ISubscriberService,ICapSubscribe
{
[CapSubscribe("xxx.services.show.time")]
public void CheckReceivedMessage(DateTime datetime)
{
}
}
}
更多信息请参考官方文档：https://cap.dotnetcore.xyz/
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
将JSON字符串修饰为友好的可读格式。 在线工具，JSON美化和格式化在线工具，online