CAP


●CAP docs看到更多详细资料。
●CAP 视频教程，学习如何在项目中集成CAP。
●GitHub源码：https://github.com/dotnetcore/cap
●示例代码：https://github.com/dotnetcore/CAP/tree/master/samples

CAP 运输器

通过运输将数据从一个地方移动到另一个地方-在采集程序和管道之间，管道与实体数据库之间，甚至在管道与外部系统之间。

怎么选择运输器

🏳‍🌈 RabbitMQ Kafka Azure Service Bus In-Memory
定位 可靠消息传输 实时数据处理 云 内存型，测试
分布式 ✔ ✔ ✔ ❌
持久化 ✔ ✔ ✔ ❌
性能 Medium High Medium High

RabbitMQ

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

RabbitMQ Options

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

NAME DESCRIPTION TYPE DEFAULT
HostName 宿主地址，配置集群可以使用逗号分隔，如 192.168.1.111,192.168.1.112 string localhost
UserName 用户名 string guest
Password 密码 string guest
VirtualHost 虚拟主机 string /
Port 端口号 int -1
ExchangeName CAP默认Exchange名称 string cap.default.topic
QueueArguments 创建队列额外参数 x-arguments QueueArgumentsOptions N/A
ConnectionFactoryOptions RabbitMQClient原生参数 ConnectionFactory N/A
CustomHeaders 订阅者自定义头信息 Func>> N/A

CustomHeaders Options

当需要从异构系统或者直接接收从RabbitMQ 控制台发送的消息时，由于 CAP 需要定义额外的头信息才能正常订阅，所以此时会出现异常。通过提供此参数来进行自定义头信息的设置来使订阅者正常工作。
你可以在这里找到有关 头信息 的说明。
用法如下：

  1. config.UseRabbitMQ(o =>
  2. {
  3.     o.CustomHeaders = e => new List<KeyValuePair<string, string>>
  4.     {
  5.         new KeyValuePair<string, string>(Headers.MessageId, SnowflakeId.Default().NextId().ToString()),
  6.         new KeyValuePair<string, string>(Headers.MessageName, e.RoutingKey),
  7.     };
  8. });

连接 RabbitMQ 集群

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

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

RabbitMQ.MySQL示例

注意,需要在mysql数据库手工创建好表

  1. CREATE TABLE `persons` (
  2.   `Id` int NOT NULL,
  3.   `name` varchar(55) DEFAULT NULL
  4. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

NuGet

  1. Install-Package DotNetCore.CAP.RabbitMQ
  2. Install-Package DotNetCore.CAP.Dashboard
  3. Install-Package DotNetCore.CAP.MySql
  4. Install-Package Pomelo.EntityFrameworkCore.MySql
  5. Install-Package Microsoft.EntityFrameworkCore.Design
  6. Install-Package Dapper

注入CAP

  3. using CAP.Transport.RabbitMQ.MySql;
  4. using CAP.Transport.RabbitMQ.MySql.Models;
  5. using DotNetCore.CAP.Internal;
  6. using DotNetCore.CAP.Messages;
  8. var builder = WebApplication.CreateBuilder(args);
  10. // Add services to the container.
  11. builder.Services.AddControllers();
  12. builder.Services.AddEndpointsApiExplorer();
  13. builder.Services.AddSwaggerGen();
  15. var appSetting = new AppSetting();
  16. builder.Configuration.GetSection("AppSetting").Bind(appSetting);
  18. //把AppSetting实体注入到容器,方便在构造函数里使用IOptionsSnapshot<AppSetting> options
  19. builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
  21. builder.Services.AddDbContext<AppDbContext>();
  23. builder.Services.AddCap(config =>
  24. {
  25.     config.UseEntityFramework<AppDbContext>();
  27.     config.UseRabbitMQ(o =>
  28.     {
  29.         o.HostName = appSetting.RabbitMQSetting.HostName;
  30.         o.UserName = appSetting.RabbitMQSetting.UserName;
  31.         o.Password = appSetting.RabbitMQSetting.Password;
  32.         o.Port = appSetting.RabbitMQSetting.Port;
  33.         o.CustomHeaders = e => new List<KeyValuePair<string, string>>
  34.         {
  35.             new KeyValuePair<string, string>(Headers.MessageId, SnowflakeId.Default().NextId().ToString()),
  36.             new KeyValuePair<string, string>(Headers.MessageName, e.RoutingKey),
  37.         };
  38.         o.ConnectionFactoryOptions = opt => {
  39.             //rabbitmq client ConnectionFactory config
  40.         };
  41.     });//配置RabbitMQ
  43.     config.UseDashboard();//配置Dashboard
  45.     config.FailedRetryCount = 5;
  46.     config.FailedThresholdCallback = failed =>
  47.     {
  48.         var logger = failed.ServiceProvider.GetService<ILogger<Program>>();
  49.         logger?.LogError($@"A message of type {failed.MessageType} failed after executing {config.FailedRetryCount} several times,requiring manual troubleshooting. Message name: {failed.Message.GetName()}");
  50.     };
  51. });
  53. var app = builder.Build();
  55. // Configure the HTTP request pipeline.
  56. if (app.Environment.IsDevelopment())
  57. {
  58.     app.UseSwagger();
  59.     app.UseSwaggerUI();
  60. }
  62. app.UseHttpsRedirection();
  64. app.UseAuthorization();
  66. app.MapControllers();
  68. app.Run();

公共部分

  1. {
  2.   "Logging": {
  3.     "LogLevel": {
  4.       "Default": "Information",
  5.       "Microsoft.AspNetCore": "Warning"
  6.     }
  7.   },
  8.   "AllowedHosts": "*",
  9.   "AppSetting": {
  10.     "SqlServerSetting": {
  11.       "Connection": "Server=192.168.3.40;Database=webdemo;User=sa;Password=longfuchu;"
  12.     },
  13.     "MysqlSetting": {
  14.       "Connection": "server=192.168.3.40;port=3306;user=root;password=123456;database=webdemo;charset=utf8;Allow Zero Datetime=true;sslmode=none;Old Guids=true;Allow User Variables=True"
  15.     },
  16.     "RedisSetting": {
  17.       "Connection": "192.168.3.40:6379,password=",
  18.       "Database": "test"
  19.     },
  20.     "RabbitMQSetting": {
  21.       "HostName": "localhost",
  22.       "UserName": "admin",
  23.       "Password": "admin",
  24.       "Port": 5672
  25.     },
  26.     "MongoDbSetting": {
  27.       "Connection": "mongodb://192.168.3.40:2717", ///?replicaSet=rs0
  28.       "Database": "test"
  29.     }
  30.   }
  31. }
  1. namespace CAP.Transport.RabbitMQ.MySql.Models;
  3. /// <summary>
  4. /// mongodb setting
  5. /// </summary>
  6. public class AppSetting
  7. {
  8.     /// <summary>
  9.     /// SqlServerSetting
  10.     /// </summary>
  11.     public SqlServerSetting SqlServerSetting { get; set; }
  12.     /// <summary>
  13.     /// MysqlSetting
  14.     /// </summary>
  15.     public MysqlSetting MysqlSetting { get; set; }
  16.     /// <summary>
  17.     /// RedisSetting
  18.     /// </summary>
  19.     public RedisSetting RedisSetting { get; set; }
  20.     /// <summary>
  21.     /// RabbitMQSetting
  22.     /// </summary>
  23.     public RabbitMQSetting RabbitMQSetting { get; set; }
  25.     /// <summary>
  26.     /// MongoDbSetting
  27.     /// </summary>
  28.     public MongoDbSetting MongoDbSetting { get; set; }
  30.     /// <summary>
  31.     /// KafkaSetting
  32.     /// </summary>
  33.     public KafkaSetting KafkaSetting { get; set; }
  34. }
  35. public class RabbitMQSetting
  36. {
  37.     /// <summary>
  38.     /// HostName
  39.     /// </summary>
  40.     public string HostName { get; set; } = "localhost";
  42.     /// <summary>
  43.     /// Password
  44.     /// </summary>
  45.     public string Password { get; set; }
  47.     /// <summary>
  48.     /// Username
  49.     /// </summary>
  50.     public string UserName { get; set; }
  52.     /// <summary>
  53.     /// The port to connect on.
  54.     /// </summary>
  55.     public int Port { get; set; } = 5672;
  56. }
  57. public class RedisSetting
  58. {
  59.     /// <summary>
  60.     /// connection string
  61.     /// </summary>
  62.     public string Connection { get; set; }
  63.     /// <summary>
  64.     /// database name
  65.     /// </summary>
  66.     public string Database { get; set; }
  67. }
  68. public class MongoDbSetting
  69. {
  70.     /// <summary>
  71.     /// connection string
  72.     /// </summary>
  73.     public string Connection { get; set; }
  74.     /// <summary>
  75.     /// database name
  76.     /// </summary>
  77.     public string Database { get; set; }
  78. }
  79. public class SqlServerSetting
  80. {
  81.     /// <summary>
  82.     /// connection string
  83.     /// </summary>
  84.     public string Connection { get; set; }
  85. }
  86. public class MysqlSetting
  87. {
  88.     /// <summary>
  89.     /// connection string
  90.     /// </summary>
  91.     public string Connection { get; set; }
  92. }
  93. public class KafkaSetting
  94. {
  95.     /// <summary>
  96.     /// Servers
  97.     /// </summary>
  98.     public string Servers { get; set; }
  99. }
  1. using CAP.Transport.RabbitMQ.MySql.Models;
  2. using Microsoft.EntityFrameworkCore;
  3. using Microsoft.Extensions.Options;
  5. namespace CAP.Transport.RabbitMQ.MySql;
  8. public class AppDbContext : DbContext
  9. {
  10.     /// <summary>
  11.     /// AppDbContext
  12.     /// builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
  13.     /// </summary>
  14.     /// <param name="options"></param>
  15.     public AppDbContext(IOptionsSnapshot<AppSetting> options)
  16.     {
  17.         ConnectionString = options != null ? options.Value.SqlServerSetting.Connection : "";
  18.     }
  19.     //配置数据库连接
  20.     public static string ConnectionString;
  22.     public DbSet<Person> Persons { get; set; }
  24.     protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
  25.     {
  26.         optionsBuilder.UseMySql(ConnectionString, ServerVersion.AutoDetect(ConnectionString));
  27.     }
  28. }
  1. using Microsoft.AspNetCore.Mvc;
  3. namespace CAP.Transport.RabbitMQDemo.Controllers;
  5. /// <summary>
  6. /// 基础控制器
  7. /// </summary>
  8. [Route("api/[area]/[controller]/[action]")]
  9. [ApiController]
  10. public abstract class BaseController : ControllerBase
  11. {
  12. }
  1. using Microsoft.AspNetCore.Mvc;
  3. namespace CAP.Transport.RabbitMQDemo.Controllers.MQ;
  5. /// <summary>
  6. /// 域控制器
  7. /// </summary>
  8. [Area("Order")]
  9. public abstract class AreaController : BaseController
  10. {
  11. }

发布消息

  1. using CAP.Transport.RabbitMQ.MySql.Models;
  2. using DotNetCore.CAP;
  3. using Microsoft.AspNetCore.Mvc;
  4. using MySqlConnector;
  5. using System;
  7. namespace CAP.Transport.RabbitMQ.MySql.Controllers.MQ;
  9. /// <summary>
  10. /// 发送消息
  11. /// </summary>
  12. public class PublishController : AreaController
  13. {
  14.     ICapPublisher _capBus;
  15.     public PublishController(ICapPublisher capBus)
  16.     {
  17.         _capBus = capBus;
  18.     }
  20.     /// <summary>
  21.     /// 发布者
  22.     /// </summary>
  23.     /// <returns></returns>
  24.     [HttpGet]
  25.     public async Task<IActionResult> WithoutTransaction()
  26.     {
  27.         Console.WriteLine("Publish send message: " + DateTime.Now);
  28.         await _capBus.PublishAsync("sample.rabbitmq.mysql", DateTime.Now);
  30.         return Ok();
  31.     }
  33.     /// <summary>
  34.     /// 发布者-MySqlConnection
  35.     /// </summary>
  36.     /// <returns></returns>
  37.     [HttpPost]
  38.     public IActionResult AdonetWithTransaction(Person person)
  39.     {
  40.         if (person == null || person.Id <= 0) return Ok();
  42.         using (var connection = new MySqlConnection(AppDbContext.ConnectionString))
  43.         {
  44.             using (var transaction = connection.BeginTransaction(_capBus, true))
  45.             {
  46.                 string sqlstr = string.Format("insert into persons(Id,name) values({0},'{1}')", person.Id, person.Name);
  47.                 using (MySqlCommand cmd = new(sqlstr, connection))
  48.                 {
  49.                     Console.WriteLine("即将执行SQL语句：   " + sqlstr);
  50.                     int resut = cmd.ExecuteNonQuery();
  51.                 }
  53.                 //for (int i = 0; i < 5; i++)
  54.                 //{
  55.                 _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);
  56.                 //}
  57.             }
  58.         }
  60.         return Ok();
  61.     }
  63.     /// <summary>
  64.     /// 发布者-EF
  65.     /// </summary>
  66.     /// <returns></returns>
  67.     [HttpPost]
  68.     public IActionResult EntityFrameworkWithTransaction([FromServices] AppDbContext dbContext, Person person)
  69.     {
  70.         using (var trans = dbContext.Database.BeginTransaction(_capBus, autoCommit: false))
  71.         {
  72.             dbContext.Persons.Add(new Person() { Id=person.Id,Name = person.Name+"ef" }) ;
  74.             for (int i = 0; i < 1; i++)
  75.             {
  76.                 _capBus.Publish("sample.rabbitmq.mysql", DateTime.Now);
  77.             }
  79.             dbContext.SaveChanges();
  81.             trans.Commit();
  82.         }
  83.         return Ok();
  84.     }
  86. }

处理消息

  1. using DotNetCore.CAP;
  2. using Microsoft.AspNetCore.Mvc;
  4. namespace CAP.Transport.RabbitMQ.MySql.Controllers.MQ;
  6. /// <summary>
  7. /// 处理消息
  8. /// </summary>
  9. [ApiController]
  10. public class ConsumerController : ControllerBase
  11. {
  12.     /// <summary>
  13.     /// 订阅消费者-MySqlConnection
  14.     /// </summary>
  15.     /// <param name="value"></param>
  16.     [NonAction]
  17.     [CapSubscribe("sample.rabbitmq.mysql")]
  18.     public void Subscriber(DateTime p)
  19.     {
  20.         Console.WriteLine($@"{DateTime.Now} Subscriber invoked, Info: {p}");
  21.     }
  23.     /// <summary>
  24.     /// 订阅消费者-EF
  25.     /// </summary>
  26.     /// <param name="p"></param>
  27.     /// <param name="header"></param>
  28.     [NonAction]
  29.     [CapSubscribe("sample.rabbitmq.mysql", Group = "group.test2")]
  30.     public void Subscriber2(DateTime p, [FromCap] CapHeader header)
  31.     {
  32.         Console.WriteLine($@"{DateTime.Now} Subscriber invoked, Info: {p}");
  33.     }
  34. }

测试

https://localhost:7285/api/Order/Publish/EntityFrameworkWithTransaction
1663665077188.png
1663665097746.png
1663665352886.png
1663665373765.png