Asp .Net Core 系列：集成 CAP + RabbitMQ + MySQL（含幂等性）

简介

官网：https://cap.dotnetcore.xyz/

CAP 是什么？

是一个 EventBus，同时也是一个在微服务或者 SOA 系统中解决分布式事务问题的一个框架。它有助于创建可扩展，可靠并且易于更改的微服务系统。

什么是 EventBus？

事件总线是一种机制，它允许不同的组件彼此通信而不彼此了解。 组件可以将事件发送到 Eventbus，而无需知道是谁来接听或有多少其他人来接听。 组件也可以侦听 Eventbus 上的事件，而无需知道谁发送了事件。 这样，组件可以相互通信而无需相互依赖。 同样，很容易替换一个组件。 只要新组件了解正在发送和接收的事件，其他组件就永远不会知道.

CAP 支持的运输器

• RabbitMQ
• Kafka
• Azure Service Bus
• Amazon SQS
• NATS
• In-Memory Queue
• Redis Streams
• Apache Pulsar

CAP 支持的持久化数据库

• SQL Server
• MySQL
• PostgreSql
• MongoDB
• In-Memory Storage

集成 CAP + RabbitMQ + MySQL

安装 CAP NuGet 包

在你的.NET Core 项目中，通过 NuGet 包管理器安装 CAP。

dotnet add package DotNetCore.CAP
dotnet add package DotNetCore.CAP.RabbitMQ
dotnet add package DotNetCore.CAP.MySql
dotnet add package DotNetCore.CAP.Dashboard #Dashboard
dotnet add package Pomelo.EntityFrameworkCore.MySql #这个之后主要用于幂等性判断,可以不要

配置 CAP

        /// <summary>/// 添加分布式事务服务/// </summary>/// <param name="services">服务集合</param>/// <param name="capSection">cap链接项</param>/// <param name="rabbitMQSection">rabbitmq配置项</param>/// <param name="expiredTime">成功消息过期时间</param>/// <returns></returns>public static IServiceCollection AddMCodeCap(this IServiceCollection services, Action<CapOptions> configure = null, string capSection = "cap", string rabbitMQSection = "rabbitmq"){var rabbitMQOptions = ServiceProviderServiceExtensions.GetRequiredService<IConfiguration>(services.BuildServiceProvider()).GetSection(rabbitMQSection).Get<RabbitMQOptions>();var logger = ServiceProviderServiceExtensions.GetRequiredService<ILogger<CapContext>>(services.BuildServiceProvider());if (rabbitMQOptions == null){throw new ArgumentNullException("rabbitmq not config.");}var capJson = ServiceProviderServiceExtensions.GetRequiredService<IConfiguration>(services.BuildServiceProvider()).GetValue<string>(capSection);if (string.IsNullOrEmpty(capJson)){throw new ArgumentException("cap未设置");}//services.AddDbContext<CapContext>(options => options.UseMySql(capJson, ServerVersion.AutoDetect(capJson)));services.AddCap(x =>{//使用RabbitMQ传输x.UseRabbitMQ(opt => { opt = rabbitMQOptions; });使用MySQL持久化x.UseMySql(capJson);//x.UseEntityFramework<CapContext>();x.UseDashboard();//成功消息的过期时间（秒）x.SucceedMessageExpiredAfter = 10 * 24 * 3600;x.FailedRetryCount = 5;//失败回调，通过企业微信，短信通知人工干预x.FailedThresholdCallback = (e) =>{if (e.MessageType == MessageType.Publish){logger.LogError("Cap发送消息失败;" + JsonExtension.Serialize(e.Message));}else if (e.MessageType == MessageType.Subscribe){logger.LogError("Cap接收消息失败;" + JsonExtension.Serialize(e.Message));}};configure?.Invoke(x);});return services;}internal class JsonExtension
{private static readonly JsonSerializerSettings _jsonSerializerSettings;internal static JsonSerializerSettings CustomSerializerSettings;static JsonExtension(){_jsonSerializerSettings = DefaultSerializerSettings;}internal static JsonSerializerSettings DefaultSerializerSettings{get{var settings = new JsonSerializerSettings();// 设置如何将日期写入JSON文本。默认值为“IsoDateFormat”//settings.DateFormatHandling = DateFormatHandling.IsoDateFormat;// 设置在序列化和反序列化期间如何处理DateTime时区。默认值为 “RoundtripKind”//settings.DateTimeZoneHandling = DateTimeZoneHandling.RoundtripKind;// 设置在序列化和反序列化期间如何处理默认值。默认值为“Include”//settings.DefaultValueHandling = DefaultValueHandling.Include;// 设置写入JSON文本时DateTime和DateTimeOffset值的格式，以及读取JSON文本时预期的日期格式。默认值为“ yyyy'-'MM'-'dd'T'HH':'mm':'ss.FFFFFFFK ”。settings.DateFormatString = "yyyy-MM-dd HH:mm:ss";// 设置在序列化和反序列化期间如何处理空值。默认值为“Include”//settings.NullValueHandling = NullValueHandling.Include;// 设置序列化程序在将.net对象序列化为JSON时使用的契约解析器settings.ContractResolver = new CamelCasePropertyNamesContractResolver();// 设置如何处理引用循环(例如，类引用自身)。默认值为“Error”。settings.ReferenceLoopHandling = ReferenceLoopHandling.Ignore;// 是否格式化文本settings.Formatting = Formatting.Indented;//支持将Enum 由默认 Number类型 转换为String//settings.SerializerSettings.Converters.Add(new StringEnumConverter());//将long类型转为string//settings.SerializerSettings.Converters.Add(new NumberConverter(NumberConverterShip.Int64));return settings;}}public static T Deserialize<T>(string json, JsonSerializerSettings serializerSettings = null){if (string.IsNullOrEmpty(json)) return default;if (serializerSettings == null) serializerSettings = _jsonSerializerSettings;//值类型和String类型if (typeof(T).IsValueType || typeof(T) == typeof(string)){return (T)Convert.ChangeType(json, typeof(T));}return JsonConvert.DeserializeObject<T>(json, CustomSerializerSettings ?? serializerSettings);}public static object Deserialize(string json, Type type, JsonSerializerSettings serializerSettings = null){if (string.IsNullOrEmpty(json)) return default;if (serializerSettings == null) serializerSettings = _jsonSerializerSettings;return JsonConvert.DeserializeObject(json,type, CustomSerializerSettings ?? serializerSettings);}public static string Serialize<T>(T obj, JsonSerializerSettings serializerSettings = null){if (obj is null) return string.Empty;if (obj is string) return obj.ToString();if (serializerSettings == null) serializerSettings = _jsonSerializerSettings;return JsonConvert.SerializeObject(obj, CustomSerializerSettings ?? serializerSettings);}
}

appsettings.json

{"cap": "Server=127.0.0.1;Port=3306;Database=spring;Uid=root;Pwd=123456;Allow User Variables=true;Pooling=true;Min Pool Size=0;Max Pool Size=100;Connection Lifetime=0;","rabbitmq": {"HostName": "127.0.0.1","Port": 5672,"UserName": "guest","Password": "guest","VirtualHost": "/"}
}

使用 CAP 发布事件

public class YourService
{private readonly ICapPublisher _capPublisher;public YourService(ICapPublisher capPublisher){_capPublisher = capPublisher;}public async Task DoSomethingAsync(){// ... 业务逻辑 ...await _capPublisher.PublishAsync("your.event.name", new YourEventData { /* ... */ },"callback.name");}
}

订阅事件

你需要实现一个事件处理器来订阅并处理事件。这通常是通过继承 ICapSubscribe 接口或使用 CAP 的[CapSubscribe]属性来实现的

public class YourEventHandler : ICapSubscribe
{[CapSubscribe("your.event.name")]public async Task Handle(YourEventData eventData){// 处理事件逻辑}
}

或者，使用特性：

[CapSubscribe("your.event.name")]
public class YourEventHandler
{public async Task Handle(YourEventData eventData){// 处理事件逻辑}
}

其它说明

配置

DefaultGroupName

GroupNamePrefix

TopicNamePrefix

Version

FailedRetryInterval *

ConsumerThreadCount *

默认值：1

消费者线程并行处理消息的线程数，当这个值大于1时，将不能保证消息执行的顺序。

FailedRetryCount *

默认值：50

重试的最大次数。当达到此设置值时，将不会再继续重试，通过改变此参数来设置重试的最大次数。

SucceedMessageExpiredAfter

FailedMessageExpiredAfter *

EnablePublishParallelSend

补偿事务

某些情况下，消费者需要返回值以告诉发布者执行结果，以便于发布者实施一些动作，通常情况下这属于补偿范围。

你可以在消费者执行的代码中通过重新发布一个新消息来通知上游，CAP 提供了一种简单的方式来做到这一点。 你可以在发送的时候指定 callbackName 来得到消费者的执行结果，通常这仅适用于点对点的消费。以下是一个示例。

例如，在一个电商程序中，订单初始状态为 pending，当商品数量成功扣除时将状态标记为 succeeded ，否则为 failed。

序列化

意味着你可以调整序列化配置

自定义序列化

 public class MessageSerializer : ISerializer{public Message Deserialize(string json){return JsonExtension.Deserialize<Message>(json);}public object Deserialize(object value, Type valueType){if (value is JToken jToken){return jToken.ToObject(valueType);}throw new NotSupportedException("Type is not of type JToken");}public ValueTask<Message> DeserializeAsync(TransportMessage transportMessage, Type valueType){if (valueType == null || transportMessage.Body.IsEmpty){return ValueTask.FromResult(new DotNetCore.CAP.Messages.Message(transportMessage.Headers, null));}var json = Encoding.UTF8.GetString(transportMessage.Body.ToArray());return ValueTask.FromResult(new DotNetCore.CAP.Messages.Message(transportMessage.Headers, JsonExtension.Deserialize(json, valueType)));}public bool IsJsonType(object jsonObject){return jsonObject is JsonToken || jsonObject is JToken;}public string Serialize(Message message){return JsonExtension.Serialize(message);}public ValueTask<TransportMessage> SerializeAsync(Message message){if (message == null){throw new ArgumentNullException(nameof(message));}if (message.Value == null){return ValueTask.FromResult(new TransportMessage(message.Headers, null));}var json = JsonExtension.Serialize(message.Value);return ValueTask.FromResult(new TransportMessage(message.Headers, Encoding.UTF8.GetBytes(json)));}}

然后将你的实现注册到容器中:

services.AddSingleton<DotNetCore.CAP.Serialization.ISerializer, MessageSerializer>();
services.AddCap(x =>
{ xxx
}

事务

CAP 不直接提供开箱即用的基于 DTC 或者 2PC 的分布式事务，相反我们提供一种可以用于解决在分布式事务遇到的问题的一种解决方案。

在分布式环境中，由于涉及通讯的开销，使用基于2PC或DTC的分布式事务将非常昂贵，在性能方面也同样如此。另外由于基于2PC或DTC的分布式事务同样受CAP定理的约束，当发生网络分区时它将不得不放弃可用性(CAP中的A)。

针对于分布式事务的处理，CAP 采用的是“异步确保”这种方案。类似于Java中Seata的Saga模式

幂等性

在说幂等性之前，我们先来说下关于消费端的消息交付。

由于CAP不是使用的 MS DTC 或其他类型的2PC分布式事务机制，所以存在至少消息严格交付一次的问题，具体的说在基于消息的系统中，存在以下三种可能：

• Exactly Once() （仅有一次）
• At Most Once （最多一次）
• At Least Once （最少一次）

在CAP中，我们采用的交付保证为 At Least Once。

由于我们具有临时存储介质（数据库表），也许可以做到 At Most Once, 但是为了严格保证消息不会丢失，我们没有提供相关功能或配置。

以自然的方式处理幂等消息

显式处理幂等消息

    public interface IMessageTracker{Task<bool> HasProcessedAsync(string msgId);bool HasProcessed(string msgId);Task MarkAsProcessedAsync(string msgId);void MarkAsProcessed(string msgId);}internal class MessageTrackLog{public MessageTrackLog(string messageId){MessageId = messageId;CreatedTime = DateTime.Now;}public string MessageId { get; set; }public DateTime CreatedTime { get; set; }}public class MessageData<T>{public string Id { get; set; }public T MessageBody { get; set; }public DateTime CreatedTime { get; set; }public MessageData(T messageBody){MessageBody = messageBody;CreatedTime = DateTime.Now;Id = SnowflakeGenerator.Instance().GetId().ToString();}}internal class SnowflakeGenerator{private static long machineId;//机器IDprivate static long datacenterId = 0L;//数据IDprivate static long sequence = 0L;//计数从零开始private static long twepoch = 687888001020L; //惟一时间随机量private static long machineIdBits = 5L; //机器码字节数private static long datacenterIdBits = 5L;//数据字节数public static long maxMachineId = -1L ^ -1L << (int)machineIdBits; //最大机器IDprivate static long maxDatacenterId = -1L ^ (-1L << (int)datacenterIdBits);//最大数据IDprivate static long sequenceBits = 12L; //计数器字节数，12个字节用来保存计数码        private static long machineIdShift = sequenceBits; //机器码数据左移位数，就是后面计数器占用的位数private static long datacenterIdShift = sequenceBits + machineIdBits;private static long timestampLeftShift = sequenceBits + machineIdBits + datacenterIdBits; //时间戳左移动位数就是机器码+计数器总字节数+数据字节数public static long sequenceMask = -1L ^ -1L << (int)sequenceBits; //一微秒内能够产生计数，若是达到该值则等到下一微妙在进行生成private static long lastTimestamp = -1L;//最后时间戳private static object syncRoot = new object();//加锁对象static SnowflakeGenerator snowflake;static SnowflakeGenerator(){snowflake = new SnowflakeGenerator();}public static SnowflakeGenerator Instance(){if (snowflake == null)snowflake = new SnowflakeGenerator();return snowflake;}public SnowflakeGenerator(){Snowflakes(0L, -1);}public SnowflakeGenerator(long machineId){Snowflakes(machineId, -1);}public SnowflakeGenerator(long machineId, long datacenterId){Snowflakes(machineId, datacenterId);}private void Snowflakes(long machineId, long datacenterId){if (machineId >= 0){if (machineId > maxMachineId){throw new Exception("机器码ID非法");}SnowflakeGenerator.machineId = machineId;}if (datacenterId >= 0){if (datacenterId > maxDatacenterId){throw new Exception("数据中心ID非法");}SnowflakeGenerator.datacenterId = datacenterId;}}/// <summary>/// 生成当前时间戳/// </summary>/// <returns>毫秒</returns>private static long GetTimestamp(){//让他2000年开始return (long)(DateTime.UtcNow - new DateTime(2000, 1, 1, 0, 0, 0, DateTimeKind.Utc)).TotalMilliseconds;}/// <summary>/// 获取下一微秒时间戳/// </summary>/// <param name="lastTimestamp"></param>/// <returns></returns>private static long GetNextTimestamp(long lastTimestamp){long timestamp = GetTimestamp();int count = 0;while (timestamp <= lastTimestamp)//这里获取新的时间,可能会有错,这算法与comb同样对机器时间的要求很严格{count++;if (count > 10)throw new Exception("机器的时间可能不对");System.Threading.Thread.Sleep(1);timestamp = GetTimestamp();}return timestamp;}/// <summary>/// 获取长整形的ID/// </summary>/// <returns></returns>public long GetId(){lock (syncRoot){long timestamp = GetTimestamp();if (SnowflakeGenerator.lastTimestamp == timestamp){ //同一微妙中生成IDsequence = (sequence + 1) & sequenceMask; //用&运算计算该微秒内产生的计数是否已经到达上限if (sequence == 0){//一微妙内产生的ID计数已达上限，等待下一微妙timestamp = GetNextTimestamp(SnowflakeGenerator.lastTimestamp);}}else{//不一样微秒生成IDsequence = 0L;}if (timestamp < lastTimestamp){throw new Exception("时间戳比上一次生成ID时时间戳还小，故异常");}SnowflakeGenerator.lastTimestamp = timestamp; //把当前时间戳保存为最后生成ID的时间戳long Id = ((timestamp - twepoch) << (int)timestampLeftShift)| (datacenterId << (int)datacenterIdShift)| (machineId << (int)machineIdShift)| sequence;return Id;}}}

基于Redis显式处理幂等消息

internal class RedisMessageTracker : IMessageTracker
{#region 属性和字段private const string KEY_PREFIX = "msgtracker:"; // 默认Key前缀private const int DEFAULT_CACHE_TIME = 60 * 60 * 24 * 3; // 默认缓存时间为3天，单位为秒private readonly IDatabase _redisDatabase;#endregion//依赖StackExchange.Redis;public RedisMessageTracker(ConnectionMultiplexer multiplexer){_redisDatabase = multiplexer.GetDatabase();}public bool HasProcessed(string msgId){return _redisDatabase.KeyExists(KEY_PREFIX + msgId);}public async Task<bool> HasProcessedAsync(string msgId){return await _redisDatabase.KeyExistsAsync(KEY_PREFIX + msgId);}public void MarkAsProcessed(string msgId){var msgRecord = new MessageTrackLog(msgId);_redisDatabase.StringSet($"{KEY_PREFIX}{msgId}", JsonExtension.Serialize(msgRecord), TimeSpan.FromMinutes(DEFAULT_CACHE_TIME));}public async Task MarkAsProcessedAsync(string msgId){var msgRecord = new MessageTrackLog(msgId);await _redisDatabase.StringSetAsync($"{KEY_PREFIX}{msgId}", JsonExtension.Serialize(msgRecord), TimeSpan.FromMinutes(DEFAULT_CACHE_TIME));}
}public static IServiceCollection AddRedisMessageTracker(this IServiceCollection services){services.AddScoped<IMessageTracker, RedisMessageTracker>();return services;}

基于Mysql显式处理幂等消息

   internal class MySqlMessageTracker : IMessageTracker{private readonly CapContext _capContext;public MySqlMessageTracker(CapContext capContext){_capContext = capContext;}public bool HasProcessed(string msgId){return _capContext.MessageTrackLogs.Any(x => x.MessageId == msgId);}public Task<bool> HasProcessedAsync(string msgId){return _capContext.MessageTrackLogs.AnyAsync(x => x.MessageId == msgId);}public void MarkAsProcessed(string msgId){MessageTrackLog messageTrackLog = new MessageTrackLog(msgId);_capContext.MessageTrackLogs.Add(messageTrackLog);_capContext.SaveChanges();}public async Task MarkAsProcessedAsync(string msgId){MessageTrackLog messageTrackLog = new MessageTrackLog(msgId);await _capContext.MessageTrackLogs.AddAsync(messageTrackLog);await _capContext.SaveChangesAsync();}}internal class CapContext : DbContext{public CapContext(DbContextOptions<CapContext> options): base(options){}protected override void OnModelCreating(ModelBuilder modelBuilder){// 可以在这里进行模型配置modelBuilder.Entity<MessageTrackLog>().ToTable("message_track_log");modelBuilder.Entity<MessageTrackLog>().HasKey(b => b.MessageId);}protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder){base.OnConfiguring(optionsBuilder);}public DbSet<MessageTrackLog> MessageTrackLogs { get; set; }}public static IServiceCollection AddMySqlMessageTracker(this IServiceCollection services){services.AddScoped<IMessageTracker, MySqlMessageTracker>();var serviceProvider = services.BuildServiceProvider();using (var context = serviceProvider.GetService<CapContext>()){context.Database.ExecuteSqlRaw(@"CREATE TABLE IF NOT EXISTS  `message_track_log` (`MessageId` varchar(255) CHARACTER SET utf8mb4 NOT NULL,`CreatedTime` datetime NOT NULL,CONSTRAINT `PK_message_track_log` PRIMARY KEY (`MessageId`)) CHARACTER SET=utf8mb4;");}return services;}

使用

    [ApiController][Route("[controller]")]public class WeatherForecastController : ControllerBase{       private readonly IMessageTracker _messageTracker;public WeatherForecastController(IMessageTracker messageTracker){_messageTracker = messageTracker;}[CapSubscribe("order.test")][NonAction]public void OrderTest(MessageData<string> messageData){try{if (_messageTracker.HasProcessed(messageData.Id))return;Console.WriteLine("业务逻辑："+messageData.MessageBody);//xxxx_messageTracker.MarkAsProcessed(messageData.Id);}catch (Exception ex){throw ex;}}}    

监控

Consul

CAP的 Dashboard 使用 Consul 作为服务发现来显示其他节点的数据，然后你就在任意节点的 Dashboard 中切换到 Servers 页面看到其他的节点。

通过点击 Switch 按钮来切换到其他的节点看到其他节点的数据，而不必访问很多地址来分别查看。

以下是一个配置示例, 你需要在每个节点分别配置：

services.AddCap(x =>
{x.UseMySql(Configuration.GetValue<string>("ConnectionString"));x.UseRabbitMQ("localhost");x.UseDashboard();x.UseConsulDiscovery(_ =>{_.DiscoveryServerHostName = "localhost";_.DiscoveryServerPort = 8500;_.CurrentNodeHostName = Configuration.GetValue<string>("ASPNETCORE_HOSTNAME");_.CurrentNodePort = Configuration.GetValue<int>("ASPNETCORE_PORT");_.NodeId = Configuration.GetValue<string>("NodeId");_.NodeName = Configuration.GetValue<string>("NodeName");});
});

启用 Dashboard

PM> Install-Package DotNetCore.CAP.Dashboard

services.AddCap(x =>
{x.UseDashboard();
});

Dashboard 配置项