C#，RabbitMQ从入门到精通，.NET8.0（路由/分布式/主题/消费重复问题 /延迟队列和死信队列/消息持久化 ）/RabbitMQ集群模式

为什么使用消息队列

消息队列（MQ）在分布式系统中用于解耦生产者和消费者，提高系统的异步处理能力、削峰填谷、增强可扩展性和可靠性。通过消息队列，任务可以异步执行，避免系统因瞬时高并发而崩溃。

消息队列场景

1. 异步处理：耗时操作（如邮件发送、日志记录）通过消息队列异步完成，避免阻塞主流程。
2. 削峰填谷：突发流量通过消息队列缓冲，避免直接冲击后端服务。
3. 解耦：系统模块间通过消息通信，降低直接依赖。
4. 最终一致性：分布式事务中通过消息队列实现数据最终一致性。

MQ的部署与实践

1. 安装RabbitMQ
   通过官方安装包或Docker部署RabbitMQ，例如：

   docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management
   

2. .NET集成
   安装NuGet包RabbitMQ.Client，初始化连接：

   var factory = new ConnectionFactory { HostName = "localhost" };
   using var connection = factory.CreateConnection();
   using var channel = connection.CreateModel();
   

发布确认和消费确认

1. 发布确认
   启用发布确认模式，确保消息成功到达Broker：

   channel.ConfirmSelect();
   channel.BasicPublish(exchange: "", routingKey: "queue", mandatory: true, basicProperties: null, body: body);
   channel.WaitForConfirmsOrDie(TimeSpan.FromSeconds(5));
   

2. 消费确认
   手动ACK确保消息被正确处理：

   var consumer = new EventingBasicConsumer(channel);
   consumer.Received += (model, ea) => {// 处理逻辑channel.BasicAck(ea.DeliveryTag, multiple: false);
   };
   channel.BasicConsume(queue: "queue", autoAck: false, consumer: consumer);
   

路由模式

通过Direct交换器实现路由键精确匹配：

channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);
channel.BasicPublish(exchange: "direct_logs", routingKey: "error", body: body);

主题模式

使用Topic交换器支持通配符匹配路由键：

channel.ExchangeDeclare(exchange: "topic_logs", type: ExchangeType.Topic);
channel.BasicPublish(exchange: "topic_logs", routingKey: "logs.error", body: body);

发布确认机制

通过事务或确认机制确保消息可靠性：

channel.TxSelect();
try {channel.BasicPublish(exchange: "", routingKey: "queue", body: body);channel.TxCommit();
} catch {channel.TxRollback();
}

消费重复问题

1. 幂等性设计：业务逻辑需支持重复消费（如唯一键约束）。
2. 去重表：记录已处理消息ID，避免重复执行。

TTL消息过期特性

设置消息或队列的TTL（Time-To-Live）：

var args = new Dictionary<string, object> { { "x-message-ttl", 60000 } };
channel.QueueDeclare(queue: "ttl_queue", arguments: args);

延迟队列和死信队列

1. 死信队列：消息过期或被拒绝时转发到死信队列：

   var args = new Dictionary<string, object> {{ "x-dead-letter-exchange", "dlx_exchange" }
   };
   channel.QueueDeclare(queue: "main_queue", arguments: args);
   

2. 延迟队列：通过TTL+死信队列模拟延迟效果。

消息持久化

确保消息和队列持久化：

var properties = channel.CreateBasicProperties();
properties.Persistent = true;
channel.QueueDeclare(queue: "persistent_queue", durable: true);
channel.BasicPublish(exchange: "", routingKey: "persistent_queue", basicProperties: properties, body: body);

RabbitMQ集群模式

1. 普通集群：节点间同步元数据，但消息不冗余。
2. 镜像队列：消息在多个节点间镜像复制，实现高可用。

   # 加入集群
   rabbitmqctl join_cluster rabbit@node1
   rabbitmqctl set_policy ha-all "^ha." '{"ha-mode":"all"}'
   

以上内容涵盖了RabbitMQ在.NET中的核心使用场景和高级特性，可根据实际需求选择配置。

推荐学习 NetCoreKevin 框架

NetCoreKevin 是一个基于 .NET Core 的开源框架，专注于简化微服务架构和身份验证的实现。它内置了 IdentityServer4 集成、JWT 认证、API 网关等功能，适合构建现代化的分布式系统。

学习资源：

• GitHub 仓库：NetCoreKevin

该框架提供了比标准 IdentityServer4 更简洁的配置方式，适合快速开发企业级应用。