CAP 是一个EventBus,同时也是一个在微服务或者SOA系统中解决分布式事务问题的一个框架。它有助于创建可扩展,可靠并且易于更改的微服务系统。
这个项目中使用到SqlServer(数据库方面大同小异)、MongoDb、Kafka、RabbitMq以及Consul
1:引入相关Nuget就行
DotNetCore.CAP
DotNetCore.CAP.Kafka
DotNetCore.CAP.RabbitMQ
DotNetCore.CAP.SqlServer
DotNetCore.CAP.MongoDB
DotNetCore.CAP.Dashboard
2:在program中使用(我的项目是基于net6的)
builder.Services.AddDbContext<AppDbContext>(opt =>
opt.UseSqlServer(builder.Configuration.GetSection("CAP:SqlServer").Value);
//Options, If you are using MongoDB
builder.Services.AddSingleton<IMongoClient>(new MongoClient(builder.Configuration.GetSection("CAP:MongoDB").Value));
builder.Services.AddCap(x =>
// If you are using EF, you need to add the configuration:
//x.UseEntityFramework<AppDbContext>(); //Options, Notice: You don't need to config x.UseSqlServer(""") again! CAP can autodiscovery.
// If you are using ADO.NET, choose to add configuration you needed:
x.UseSqlServer(builder.Configuration.GetSection("CAP:SqlServer").Value);
//x.UseMySql("Your ConnectionStrings");
//x.UsePostgreSql("Your ConnectionStrings");
// If you are using MongoDB, you need to add the configuration:
//x.UseMongoDB(opt => {
// opt.DatabaseConnection = builder.Configuration.GetSection("CAP:MongoDB").Value;
//}); //注意,仅支持MongoDB 4.0+集群
// CAP support RabbitMQ,Kafka,AzureService as the MQ, choose to add configuration you needed:
x.UseRabbitMQ(opt =>
opt.HostName = "192.168.1.7";
opt.Port = 5672;
opt.UserName = "admin";
opt.Password = "admin";
opt.VirtualHost = "/";
x.UseKafka(opt =>
opt.Servers = "192.168.1.12:9092";
//x.UseRabbitMQ(builder.Configuration.GetSection("CAP:RabbitMQ").Value);
x.UseDashboard();
DiscoveryOptions discoveryOptions = new DiscoveryOptions();
discoveryOptions.CurrentNodePort = 5173;
builder.Configuration.Bind(discoveryOptions);
// Register to Consul
x.UseDiscovery(d =>
d.DiscoveryServerHostName = "localhost";
d.DiscoveryServerPort = 8500;
d.CurrentNodeHostName = "localhost";
d.CurrentNodePort = 5222;
d.NodeId = "1";
d.NodeName = "fanlin";
d.Scheme = "http";
d.MatchPath = "/api/HealthCheck";
x.FailedRetryInterval = 10;//失败重试的间隔时间
x.FailedRetryCount = 10;//失败重试的次数
x.FailedThresholdCallback = info =>
Console.WriteLine("Publish Message Error::" + info.Message);
//x.UseKafka("ConnectionString");
//x.UseAzureServiceBus("ConnectionString");
//x.UseAmazonSQS();
3:新建PublishController,主要用于发布,代码很简单。代码如下
/// <summary>
/// CAP的消费是自动消费的
/// </summary>
[Route("api/[controller]")]
[ApiController]
public class PublishController : ControllerBase
private static string _publishName = "FanlinCAPDemo.Servces.Test";
private readonly ICapPublisher _capBus;
private readonly IConfiguration _configuration;
private readonly AppDbContext _appDbContext;
public PublishController(ICapPublisher capPublisher, IConfiguration configuration, AppDbContext appDbContext)
_capBus = capPublisher;
_configuration = configuration;
_appDbContext = appDbContext;
[HttpGet]
[Route("no/transaction")]//根目录
public async Task<IActionResult> WithoutTransaction()
Console.WriteLine("普通----无事务");
var user = _appDbContext.Users.Find("1");
await _capBus.PublishAsync(_publishName, user);//应该把数据写到publish表
return Ok();
[HttpGet]
[Route("adonet/transaction")]
public IActionResult AdonetWithTransaction()
Console.WriteLine("普通事务---连接字符串----事务");
var user = _appDbContext.Users.Find("1");
using (var connection = new SqlConnection(_configuration.GetSection("CAP:SqlServer").Value))
using (var transaction = connection.BeginTransaction(_capBus, true))
//your business logic code
_capBus.Publish(_publishName, user);
return Ok();
[HttpGet]
[Route("ef/transaction")]
public IActionResult EntityFrameworkWithTransaction()
Console.WriteLine("上下文---事务");
var user = _appDbContext.Users.Find("1");
//带header
IDictionary<string, string?> dicHeader = new Dictionary<string, string?>();
dicHeader.Add("Husband", "Fanlin");
dicHeader.Add("Wife", "Baoting");
dicHeader.Add("SumAge", "34");
using (var trans = _appDbContext.Database.BeginTransaction(_capBus, autoCommit: true))
//your business logic code
_capBus.Publish(_publishName, user, dicHeader);
return Ok();
4:新建ConsumerController 用于消费,代码很简单,需要注意,订阅需要标注特性CapSubscribe,其属性name就是发布者的_publishName,也可以顶一个多个group,在rabbitmq里面会自动新建相应的队列!代码如下
[Route("api/[controller]")]
[ApiController]
public class ConsumerController : ControllerBase
private readonly AppDbContext _context;
private readonly IMongoClient _client;
private readonly ICapPublisher _capBus;
public ConsumerController(AppDbContext appDbContext, IMongoClient mongoClient, ICapPublisher iCapPublisher)
_context = appDbContext;
_capBus = iCapPublisher;
_client = mongoClient;
var collection = _client.GetDatabase("MyCap").GetCollection<Users>("MyCap.User");
collection.InsertOne(new Users()
UserID = "3",
UserName = "范小包",
UserPy = "fxb",
UserPwd = "1234",
department = "1",
userRemark = "jkl",
UserWb = "abc",
headImageFile = "none"
[NonAction]
[CapSubscribe("FanlinCAPDemo.Servces.Test")]
public void CheckReceivedMessage(Users users, [FromCap] CapHeader header)
Console.WriteLine($"{DateTime.Now} NoGroup Info:{users},Header:{header.Count}");
[NonAction]
[CapSubscribe("FanlinCAPDemo.Servces.Test", Group = "Group1")]
public void CheckReceivedMessageGroup(Users users)
Console.WriteLine($"{DateTime.Now} NoGroup Info:{users}");
//这里不支持事务 使用事务会报错
//using (var session = _client.StartTransaction(_capBus, autoCommit: false))
var collection = _client.GetDatabase("test").GetCollection<BsonDocument>("test.collection");
collection.InsertOne( new BsonDocument { { "hello", "world" } });
_capBus.Publish("sample.rabbitmq.mongodb", DateTime.Now);
//session.CommitTransaction();
5:调用接口,查看kafka和rabbitmq以及相关数据库的变化
a:RabbitMQ:
这里自动新建了两个队列,如果有定义group已group属性为准,如果没有,名称是自定义的
public void ConfigureServices(IServiceCollection services)
services.AddControllers();//配置WebAPI服务
services.AddTransient&l..
CAP是一款基于.net标准的库,该库是处理分布式事务的解决方案;它具有轻量级的、开源、易于使用、具有EventBus(事件总线)功能、持久化等特点。官方中文文档:https://cap.dotnetcore.xyz/user-guide/zh/getting-started/quick-start/
基础核心Nuget包:DotNetCore.CAP
EventBus(事件总线)Nuget包:DotNetCore.CAP.RabbitMQ、DotNetCore.CAP.Kafka、DotNetCore
Install-package DotNetCore.CAP
Install-package DotNetCore.CAP.RabbitMQ
Install-package DotNetCore.CAP.SqlServer
使用方法
Startup->ConfigureServices中添加服务:
//add CAP
services.AddCap(x =>
本地消息表模式本地消息表模式,其作为柔性事务的一种,核心是将一个分布式事务拆分为多个本地事务,事务之间通过事件消息衔接,事件消息和上个事务共用一个本地事务存储到本地消息表,再通过定时任务轮询本地消息表进行消息投递,下游业务订阅消息进行消费,本质上是依靠消息的重试机制达到最终一致性。其示意图如下所示,主要分为以下三步:本地业务数据和发布的事件消息共享同一个本地事务,进行数据落库,其中事件消息持久化到...
CAP 是一个遵循 .NET Standard 标准库的C#库,用来处理分布式事务以及提供EventBus的功能,她具有轻量级,高性能,易使用等特点。
目前 CAP 使用的是 .NET Standard 1.6 的标准进行开发,目前最新预览版本已经支持 .NET Standard 2.0
## CAP 的应用场景主要有以下两个:
MITCAP 是一个在分布式系统中(SOA,MicroService)实现事件总线及最终一致性(分布式事务)的一个开源的 C# 库,她具有轻量级,高性能,易使用等特点。你可以轻松的在基于 .NET Core 技术的分布式系统中引入CAP,包括但限于 ASP.NET Core 和 ASP.NET Core on .NET Framework。CAP 以 NuGet 包的形式提供,对项目无任何入侵,你仍然可以以你喜爱的方式来构建分布式系统。