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.SqlServer示例

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

  1. CREATE TABLE [dbo].[persons](
  2.     [Id] [int] NOT NULL,
  3.     [name] [varchar](55) NULL
  4. ) ON [PRIMARY]
  5. GO

NuGet

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

注入CAP

  1. using CAP.Transport.RabbitMQ.SqlServer;
  2. using CAP.Transport.RabbitMQ.SqlServer.Models;
  3. using DotNetCore.CAP.Internal;
  4. using DotNetCore.CAP.Messages;
  6. var builder = WebApplication.CreateBuilder(args);
  8. // Add services to the container.
  9. builder.Services.AddControllers();
  10. builder.Services.AddEndpointsApiExplorer();
  11. builder.Services.AddSwaggerGen();
  13. var appSetting = new AppSetting();
  14. builder.Configuration.GetSection("AppSetting").Bind(appSetting);
  16. //把AppSetting实体注入到容器,方便在构造函数里使用IOptionsSnapshot<AppSetting> options
  17. builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
  19. //[FromServices] AppDbContext dbContext
  20. builder.Services.AddDbContext<AppDbContext>();
  22. builder.Services.AddCap(config =>
  23. {
  24.     config.UseEntityFramework<AppDbContext>();
  26.     config.UseRabbitMQ(o =>
  27.     {
  28.         o.HostName = appSetting.RabbitMQSetting.HostName;
  29.         o.UserName = appSetting.RabbitMQSetting.UserName;
  30.         o.Password = appSetting.RabbitMQSetting.Password;
  31.         o.Port = appSetting.RabbitMQSetting.Port;
  32.         o.CustomHeaders = e => new List<KeyValuePair<string, string>>
  33.         {
  34.             new KeyValuePair<string, string>(Headers.MessageId, SnowflakeId.Default().NextId().ToString()),
  35.             new KeyValuePair<string, string>(Headers.MessageName, e.RoutingKey),
  36.         };
  37.         o.ConnectionFactoryOptions = opt => {
  38.             //rabbitmq client ConnectionFactory config
  39.         };
  40.     });//配置RabbitMQ
  42.     config.UseDashboard();//配置Dashboard
  44.     config.FailedRetryCount = 5;
  45.     config.FailedThresholdCallback = failed =>
  46.     {
  47.         var logger = failed.ServiceProvider.GetService<ILogger<Program>>();
  48.         logger?.LogError($@"A message of type {failed.MessageType} failed after executing {config.FailedRetryCount} several times,requiring manual troubleshooting. Message name: {failed.Message.GetName()}");
  49.     };
  50. });
  52. var app = builder.Build();
  54. // Configure the HTTP request pipeline.
  55. if (app.Environment.IsDevelopment())
  56. {
  57.     app.UseSwagger();
  58.     app.UseSwaggerUI();
  59. }
  61. app.UseHttpsRedirection();
  63. app.UseAuthorization();
  65. app.MapControllers();
  67. 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.SqlServer.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; }
  29.     /// <summary>
  30.     /// KafkaSetting
  31.     /// </summary>
  32.     public KafkaSetting KafkaSetting { get; set; }
  33. }
  34. public class RabbitMQSetting
  35. {
  36.     /// <summary>
  37.     /// HostName
  38.     /// </summary>
  39.     public string HostName { get; set; } = "localhost";
  41.     /// <summary>
  42.     /// Password
  43.     /// </summary>
  44.     public string Password { get; set; }
  46.     /// <summary>
  47.     /// Username
  48.     /// </summary>
  49.     public string UserName { get; set; }
  51.     /// <summary>
  52.     /// The port to connect on.
  53.     /// </summary>
  54.     public int Port { get; set; } = 5672;
  55. }
  56. public class RedisSetting
  57. {
  58.     /// <summary>
  59.     /// connection string
  60.     /// </summary>
  61.     public string Connection { get; set; }
  62.     /// <summary>
  63.     /// database name
  64.     /// </summary>
  65.     public string Database { get; set; }
  66. }
  67. public class MongoDbSetting
  68. {
  69.     /// <summary>
  70.     /// connection string
  71.     /// </summary>
  72.     public string Connection { get; set; }
  73.     /// <summary>
  74.     /// database name
  75.     /// </summary>
  76.     public string Database { get; set; }
  77. }
  78. public class SqlServerSetting
  79. {
  80.     /// <summary>
  81.     /// connection string
  82.     /// </summary>
  83.     public string Connection { get; set; }
  84. }
  85. public class MysqlSetting
  86. {
  87.     /// <summary>
  88.     /// connection string
  89.     /// </summary>
  90.     public string Connection { get; set; }
  91. }
  92. public class KafkaSetting
  93. {
  94.     /// <summary>
  95.     /// Servers
  96.     /// </summary>
  97.     public string Servers { get; set; }
  98. }
  1. using CAP.Transport.RabbitMQ.SqlServer.Models;
  2. using DotNetCore.CAP;
  3. using Microsoft.EntityFrameworkCore;
  4. using Microsoft.Extensions.Configuration;
  5. using Microsoft.Extensions.Options;
  7. namespace CAP.Transport.RabbitMQ.SqlServer;
  9. public class AppDbContext : DbContext
  10. {
  11.     /// <summary>
  12.     /// AppDbContext
  13.     /// builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
  14.     /// </summary>
  15.     /// <param name="options"></param>
  16.     public AppDbContext(IOptionsSnapshot<AppSetting> options)
  17.     {
  18.         ConnectionString = options != null ? options.Value.SqlServerSetting.Connection : "";
  19.     }
  20.     public static string ConnectionString;
  22.     public DbSet<Person> Persons { get; set; }
  24.     protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
  25.     {
  26.         optionsBuilder.UseSqlServer(ConnectionString);
  27.     }
  28. }
  1. using Microsoft.AspNetCore.Mvc;
  3. namespace CAP.Transport.RabbitMQ.SqlServer.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.RabbitMQ.SqlServer.Controllers.MQ;
  5. /// <summary>
  6. /// 域控制器
  7. /// </summary>
  8. [Area("MQ")]
  9. public abstract class AreaController : BaseController
  10. {
  11. }

发布消息

  2. using CAP.Transport.RabbitMQ.SqlServer.Models;
  3. using DotNetCore.CAP;
  4. using Microsoft.AspNetCore.Mvc;
  5. using Microsoft.Data.SqlClient;
  7. namespace CAP.Transport.RabbitMQ.SqlServer.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.         await _capBus.PublishAsync("sample.rabbitmq.sqlserver", new Person()
  28.         {
  29.             Id = 123,
  30.             Name = "Bar"
  31.         });
  33.         return Ok();
  34.     }
  36.     /// <summary>
  37.     /// 发布者-SqlConnection
  38.     /// </summary>
  39.     /// <returns></returns>
  40.     [HttpPost]
  41.     public IActionResult AdonetWithTransaction(Person person)
  42.     {
  43.         using (var connection = new SqlConnection(AppDbContext.ConnectionString))
  44.         {
  45.             using (var transaction = connection.BeginTransaction(_capBus, true))
  46.             {
  47.                 //your business code
  48.                 string sqlstr = string.Format("insert into persons(Id,name) values({0},'{1}')", person.Id, person.Name);
  49.                 using (SqlCommand cmd = new(sqlstr, connection))
  50.                 {
  51.                     Console.WriteLine("即将执行SQL语句：   " + sqlstr);
  52.                     int resut = cmd.ExecuteNonQuery();
  53.                 }
  55.                 _capBus.Publish("sample.rabbitmq.sqlserver", new Person()
  56.                 {
  57.                     Id = 123,
  58.                     Name = "Bar"
  59.                 });
  62.             }
  63.         }
  65.         return Ok();
  66.     }
  68.     /// <summary>
  69.     /// 发布者-EF
  70.     /// </summary>
  71.     /// <returns></returns>
  72.     [HttpPost]
  73.     public IActionResult EntityFrameworkWithTransaction([FromServices] AppDbContext dbContext, Person person)
  74.     {
  75.         using (dbContext.Database.BeginTransaction(_capBus, autoCommit: true))
  76.         {
  77.             dbContext.Persons.Add(new Person() { Id = person.Id, Name = person.Name + "ef" });
  78.             dbContext.SaveChanges();
  79.             _capBus.Publish("sample.rabbitmq.sqlserver", new Person()
  80.             {
  81.                 Id = 123,
  82.                 Name = "Bar"
  83.             });
  84.         }
  85.         return Ok();
  86.     }
  87. }

处理消息

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

测试