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

Redis Streams

Redis 是一个开源（BSD许可）的，内存中的数据结构存储系统，它可以用作数据库、缓存和消息中间件。
Redis Stream 是 Redis 5.0 引入的一种新数据类型，它用一种仅附加的数据结构以更抽象的方式模拟日志数据结构。
Redis Streams 可以在 CAP 中用作消息传输器。

Redis Streams Options

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

NAME DESCRIPTION TYPE DEFAULT
Configuration redis连接配置 (StackExchange.Redis) ConfigurationOptions ConfigurationOptions
StreamEntriesCount 读取时从 stream 返回的条目数 uint 10
ConnectionPoolSize 连接池数 uint 10

Redis ConfigurationOptions

如果需要更多原生Redis相关配置选项，您可以在 Configuration 选项中进行设置 :

  1. services.AddCap(capOptions => 
  2. {
  3.     capOptions.UseRedis(redisOptions=>
  4.     {
  5.         // redis options.
  6.         redisOptions.Configuration.EndPoints.Add(IPAddress.Loopback, 0);
  7.     });
  8. });

Redis.SqlServer示例

  • 已发布消息和接收的消息存储在SqlServer里
  • 订阅方法的主题名称或交换路由器密钥存储在Redis

    NuGet

    1. Install-Package DotNetCore.CAP.Dashboard
    2. Install-Package DotNetCore.CAP.RedisStreams
    3. Install-Package DotNetCore.CAP.SqlServer
    4. Install-Package Microsoft.VisualStudio.Azure.Containers.Tools.Targets

    注入CAP

    ```csharp

//已发布消息和接收的消息存储在SqlServer里 //订阅方法的主题名称或交换路由器密钥存储在Redis using CAP.Redis.SqlServer.Models; using DotNetCore.CAP.Messages;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.

builder.Services.AddControllers(); builder.Services.AddEndpointsApiExplorer(); builder.Services.AddSwaggerGen();

var appSetting = new AppSetting(); builder.Configuration.GetSection(“AppSetting”).Bind(appSetting);

//把AppSetting实体注入到容器,方便在构造函数里使用IOptionsSnapshot options builder.Services.Configure(builder.Configuration.GetSection(“AppSetting”));

builder.Services.AddCap(config => { config.UseRedis(appSetting.RedisSetting.Connection);

  1. config.UseSqlServer(appSetting.SqlServerSetting.Connection);
  3. config.UseDashboard();//配置Dashboard
  5. config.FailedRetryCount = 5;
  6. config.FailedThresholdCallback = failed =>
  7. {
  8.     var logger = failed.ServiceProvider.GetService<ILogger<Program>>();
  9.     logger?.LogError($@"A message of type {failed.MessageType} failed after executing {config.FailedRetryCount} several times,requiring manual troubleshooting. Message name: {failed.Message.GetName()}");
  10. };

});

var app = builder.Build();

// Configure the HTTP request pipeline. if (app.Environment.IsDevelopment()) { app.UseSwagger(); app.UseSwaggerUI(); }

app.UseAuthorization();

app.MapControllers();

app.Run();

  1. <a name="JCxZD"></a>
  2. ### 公共部分
  3. ```csharp
  4. {
  5.   "Logging": {
  6.     "LogLevel": {
  7.       "Default": "Information",
  8.       "Microsoft.AspNetCore": "Warning"
  9.     }
  10.   },
  11.   "AllowedHosts": "*",
  12.   "AppSetting": {
  13.     "SqlServerSetting": {
  14.       "Connection": "Server=192.168.3.40;Database=webdemo;User=sa;Password=longfuchu;"
  15.     },
  16.     "MysqlSetting": {
  17.       "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"
  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.   "MongoDbSetting": {
  30.     "Connection": "mongodb://192.168.3.40:2717", ///?replicaSet=rs0
  31.     "Database": "test"
  32.   },
  33.   "KafkaSetting": {
  34.     "Servers": "kafka1:9091,kafka2:9092,kafka3:9093"
  35.   }
  36.   }
  37. }
  1. namespace CAP.Redis.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; }
  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. namespace CAP.Redis.SqlServer.Models;
  3. public class Person
  4. {
  5.     public string Name { get; set; }
  7.     public int Age { get; set; }
  9.     public override string ToString()
  10.     {
  11.         return "Name:" + Name + ", Age:" + Age;
  12.     }
  13. }
  1. using Microsoft.AspNetCore.Mvc;
  3. namespace CAP.Redis.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.Redis.SqlServer.Controllers.Redis;
  5. /// <summary>
  6. /// 域控制器
  7. /// </summary>
  8. [Area("Redis")]
  9. public abstract class AreaController : BaseController
  10. {
  11. }

发布消息

  1. using CAP.Redis.SqlServer.Models;
  2. using DotNetCore.CAP;
  3. using Microsoft.AspNetCore.Mvc;
  5. namespace CAP.Redis.SqlServer.Controllers.Redis;
  7. /// <summary>
  8. /// 发送消息
  9. /// </summary>
  10. public class PublishController : AreaController
  11. {
  12.     private readonly ICapPublisher _capBus;
  13.     private readonly ILogger<PublishController> _logger;
  14.     public PublishController(ILogger<PublishController> logger, ICapPublisher capBus)
  15.     {
  16.         _logger = logger;
  17.         _capBus = capBus;
  18.     }
  19.     /// <summary>
  20.     /// 发布者
  21.     /// </summary>
  22.     /// <param name="message"></param>
  23.     /// <returns></returns>
  24.     [HttpGet]
  25.     public async Task Publish([FromQuery] string message = "test-message")
  26.     {
  27.         await _capBus.PublishAsync(message, new Person() { Age = 11, Name = "James" });
  28.     }
  29. }

处理消息

  1. using CAP.Redis.SqlServer.Models;
  2. using DotNetCore.CAP;
  3. using DotNetCore.CAP.Messages;
  4. using Microsoft.AspNetCore.Mvc;
  6. namespace CAP.Redis.SqlServer.Controllers.Redis;
  8. /// <summary>
  9. /// 处理消息
  10. /// </summary>
  11. [ApiController]
  12. public class ConsumerController : ControllerBase
  13. {
  14.     private readonly ILogger<ConsumerController> _logger;
  15.     public ConsumerController(ILogger<ConsumerController> logger)
  16.     {
  17.         _logger = logger;
  18.     }
  19.     /// <summary>
  20.     /// 订阅消费者-
  21.     /// </summary>
  22.     /// <param name="value"></param>
  23.     [NonAction]
  24.     [CapSubscribe("test-message")]
  25.     [CapSubscribe("test-message-1")]
  26.     [CapSubscribe("test-message-2")]
  27.     [CapSubscribe("test-message-3")]
  28.     public void Subscribe(Person p, [FromCap] CapHeader header)
  29.     {
  30.         _logger.LogInformation($"{header[Headers.MessageName]} subscribed with value --> " + p);
  31.     }
  32. }

测试

1663729555932.png
1663729594344.png