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

Apache Kafka®

Apache Kafka® 是一个开源流处理软件平台，由 LinkedIn 开发并捐赠给 Apache Software Foundation，用 Scala 和 Java 编写。
CAP 支持使用 Apache Kafka® 作为消息传输器。

Kafka Option

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

NAME DESCRIPTION TYPE DEFAULT
Servers Broker 地址 string
ConnectionPoolSize 用户名 int 10
CustomHeaders 设置自定义头 Function

Kafka示例

NuGet

  1. Install-Package DotNetCore.CAP.Kafka
  2. Install-Package DotNetCore.CAP.Dashboard
  3. Install-Package DotNetCore.CAP.PostgreSql

注入CAP

  1. using CAP.Transport.Kafka.PostgreSql.Models;
  3. var builder = WebApplication.CreateBuilder(args);
  5. // Add services to the container.
  7. builder.Services.AddControllers();
  8. // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
  9. builder.Services.AddEndpointsApiExplorer();
  10. builder.Services.AddSwaggerGen();
  12. var appSetting = new AppSetting();
  13. builder.Configuration.GetSection("AppSetting").Bind(appSetting);
  15. //把AppSetting实体注入到容器,方便在构造函数里使用IOptionsSnapshot<AppSetting> options
  16. builder.Services.Configure<AppSetting>(builder.Configuration.GetSection("AppSetting"));
  18. builder.Services.AddCap(config =>
  19. {
  20.     config.UsePostgreSql(opt => { 
  21.         opt.ConnectionString = appSetting.PostgreSqlSetting.Connection; 
  22.     }); //配置一个PostgreSql
  24.     config.UseKafka(opt => {
  25.         opt.Servers = appSetting.KafkaSetting.Servers;
  26.         //KafkaOptions
  27.     });//添加基于 Kafka 的配置项
  29.     config.UseDashboard();//配置一个Dashboard
  30. });
  32. var app = builder.Build();
  34. // Configure the HTTP request pipeline.
  35. if (app.Environment.IsDevelopment())
  36. {
  37.     app.UseSwagger();
  38.     app.UseSwaggerUI();
  39. }
  41. app.UseHttpsRedirection();
  43. app.UseAuthorization();
  45. app.MapControllers();
  47. 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.     "PostgreSqlSetting": {
  17.       "Connection": "HOST=192.168.3.40;PORT=5432;DATABASE=test_db;USER ID=postgres;PASSWORD=123456;Pooling = false"
  18.     },
  19.     "RedisSetting": {
  20.       "Connection": "192.168.3.40:6379,password=",
  21.       "Database": "test"
  22.     },
  23.     "RabbitMQSetting": {
  24.       "HostName": "localhost",
  25.       "UserName": "admin",
  26.       "Password": "admin",
  27.       "Port": 5672
  28.     }
  29.   },
  30.   "MongoDbSetting": {
  31.     "Connection": "mongodb://192.168.3.40:2717", ///?replicaSet=rs0
  32.     "Database": "test"
  33.   },
  34.   "KafkaSetting": {
  35.     "Servers": "kafka1:9091,kafka2:9092,kafka3:9093"
  36.   }
  37. }
  1. namespace CAP.Transport.Kafka.PostgreSql.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.     /// PostgreSqlSetting
  18.     /// </summary>
  19.     public PostgreSqlSetting PostgreSqlSetting { get; set; }
  20.     /// <summary>
  21.     /// RedisSetting
  22.     /// </summary>
  23.     public RedisSetting RedisSetting { get; set; }
  24.     /// <summary>
  25.     /// RabbitMQSetting
  26.     /// </summary>
  27.     public RabbitMQSetting RabbitMQSetting { get; set; }
  29.     /// <summary>
  30.     /// MongoDbSetting
  31.     /// </summary>
  32.     public MongoDbSetting MongoDbSetting { get; set; }
  34.     /// <summary>
  35.     /// KafkaSetting
  36.     /// </summary>
  37.     public KafkaSetting KafkaSetting { get; set; }
  38. }
  39. public class RabbitMQSetting
  40. {
  41.     /// <summary>
  42.     /// HostName
  43.     /// </summary>
  44.     public string HostName { get; set; } = "localhost";
  46.     /// <summary>
  47.     /// Password
  48.     /// </summary>
  49.     public string Password { get; set; }
  51.     /// <summary>
  52.     /// Username
  53.     /// </summary>
  54.     public string UserName { get; set; }
  56.     /// <summary>
  57.     /// The port to connect on.
  58.     /// </summary>
  59.     public int Port { get; set; } = 5672;
  60. }
  61. public class RedisSetting
  62. {
  63.     /// <summary>
  64.     /// connection string
  65.     /// </summary>
  66.     public string Connection { get; set; }
  67.     /// <summary>
  68.     /// database name
  69.     /// </summary>
  70.     public string Database { get; set; }
  71. }
  72. public class MongoDbSetting
  73. {
  74.     /// <summary>
  75.     /// connection string
  76.     /// </summary>
  77.     public string Connection { get; set; }
  78.     /// <summary>
  79.     /// database name
  80.     /// </summary>
  81.     public string Database { get; set; }
  82. }
  83. public class SqlServerSetting
  84. {
  85.     /// <summary>
  86.     /// connection string
  87.     /// </summary>
  88.     public string Connection { get; set; }
  89. }
  90. public class MysqlSetting
  91. {
  92.     /// <summary>
  93.     /// connection string
  94.     /// </summary>
  95.     public string Connection { get; set; }
  96. }
  97. public class PostgreSqlSetting
  98. {
  99.     /// <summary>
  100.     /// connection string
  101.     /// </summary>
  102.     public string Connection { get; set; }
  103. }
  104. public class KafkaSetting
  105. {
  106.     /// <summary>
  107.     /// Servers
  108.     /// </summary>
  109.     public string Servers { get; set; }
  110. }
  1. using Microsoft.AspNetCore.Mvc;
  3. namespace CAP.Transport.KafkaDemo.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.KafkaDemo.Controllers.Kafka;
  5. /// <summary>
  6. /// 域控制器
  7. /// </summary>
  8. [Area("Order")]
  9. public abstract class AreaController : BaseController
  10. {
  11. }

发布消息

  1. using CAP.Transport.Kafka.PostgreSql.Models;
  2. using DotNetCore.CAP;
  3. using Microsoft.AspNetCore.Mvc;
  4. using Microsoft.Extensions.Options;
  5. using Npgsql;
  6. using System.Text.Json;
  8. namespace CAP.Transport.Kafka.PostgreSql.Controllers.Kafka;
  10. /// <summary>
  11. /// 发送消息
  12. /// </summary>
  13. public class PublishController : AreaController
  14. {
  15.     private readonly ICapPublisher _capBus;
  16.     private readonly string _connectionString;
  17.     public PublishController(ICapPublisher capBus, IOptionsSnapshot<AppSetting> options)
  18.     {
  19.         _capBus = capBus;
  20.         _connectionString = options == null ? options.Value.PostgreSqlSetting.Connection : "";
  21.     }
  23.     /// <summary>
  24.     /// 发布者
  25.     /// </summary>
  26.     /// <returns></returns>
  27.     [HttpGet]//表示控制器方法不是动作方法
  28.     //[Route("~/without/transaction")]
  29.     public async Task<IActionResult> WithoutTransaction()
  30.     {
  31.         Console.WriteLine("Publish send message: " + DateTime.Now);
  32.         await _capBus.PublishAsync("sample.kafka.postgrsql", DateTime.Now);
  34.         return Ok();
  35.     }
  37.     /// <summary>
  38.     /// 发布者
  39.     /// </summary>
  40.     /// <returns></returns>
  41.     [HttpPost]
  42.     //[Route("~/adonet/transaction")]
  43.     public IActionResult AdonetWithTransaction()
  44.     {
  45.         using (var connection = new NpgsqlConnection(_connectionString))
  46.         {
  47.             using var transaction = connection.BeginTransaction(_capBus, autoCommit: false);
  48.             string sqlstr = "insert into test(cname) values('test')";
  49.             using (NpgsqlCommand cmd = new(sqlstr, connection))
  50.             {
  51.                 Console.WriteLine("即将执行SQL语句：   " + sqlstr);
  52.                 int resut = cmd.ExecuteNonQuery();
  53.             }
  54.             _capBus.Publish("sample.kafka.postgrsql", DateTime.Now);
  55.             transaction.Commit();
  56.         }
  58.         return Ok();
  59.     }
  60. }

处理消息

  1. using DotNetCore.CAP;
  2. using Microsoft.AspNetCore.Mvc;
  4. namespace CAP.Transport.Kafka.PostgreSql.Controllers.Kafka;
  6. /// <summary>
  7. /// 处理消息
  8. /// </summary>
  9. [ApiController]
  10. public class ConsumerController : ControllerBase
  11. {
  12.     /// <summary>
  13.     /// 订阅消费者
  14.     /// </summary>
  15.     /// <param name="value"></param>
  16.     [NonAction]//表示控制器方法不是动作方法
  17.     [CapSubscribe("sample.kafka.postgrsql")]
  18.     public void TestKafka(DateTime value)
  19.     {
  20.         Console.WriteLine("Subscriber output message: " + value);
  21.     }
  22. }

测试

https://localhost:7285/api/Order/Publish/WithoutTransaction
1663651247201.png
1663651271349.png

https://localhost:7285/api/Order/Publish/AdonetWithTransaction
1663651341057.png
1663651322436.png