SlimMessageBus对接RabbitMQ完全指南队列、死信与重试错误处理实战【免费下载链接】SlimMessageBusLightweight message bus interface for .NET (pub/sub and request-response) with transport plugins for popular message brokers.项目地址: https://gitcode.com/gh_mirrors/sl/SlimMessageBusSlimMessageBus是一个轻量级的 .NET 消息总线message bus支持发布/订阅pub/sub和请求/响应request-response两种通信模式。搭配 RabbitMQ 传输插件后你可以用统一的 API 对接 RabbitMQ 的 Exchange、Queue 和 Binding并轻松实现死信队列DLX与重试错误处理是 .NET 微服务架构中对接 RabbitMQ 的简单可靠方案。 5分钟搭建 RabbitMQ 环境官方仓库自带 Docker Compose 编排文件其中包含带管理界面的 RabbitMQ 服务端口 5672管理界面 15672默认账号 guest/guestrabbitmq: container_name: slim.rabbitmq image: rabbitmq:4.2.3-management-alpine ports: - 5672:5672 - 15672:15672参考文件src/Infrastructure/docker-compose.yml在 NuGet 中安装SlimMessageBus.Host.RabbitMQ和SlimMessageBus.Host.Serialization.Json两个包即可开始。对接核心一行配置连接 RabbitMQRabbitMQ 模型围绕三个概念Exchange交换机生产者发消息的入口、Queue队列消费者的收件箱、Binding绑定规则决定消息如何路由到队列。SlimMessageBus 通过WithProviderRabbitMQ配置连接并自动完成拓扑交换、队列、绑定的创建无需手动写 AMQP 代码services.AddSlimMessageBus(mbb { mbb.WithProviderRabbitMQ(cfg cfg.ConnectionString configuration[RabbitMQ:ConnectionString]); mbb.ProduceOrderEvent(x x.Exchange(orders, exchangeType: ExchangeType.Fanout)); mbb.ConsumeOrderEvent(x x .Queue(orders-queue) .ExchangeBinding(orders) .DeadLetterExchange(orders-dlq, exchangeType: ExchangeType.Direct) .WithConsumerOrderCreatedConsumer()); mbb.AddJsonSerializer(); }); 关键点SlimMessageBus 会自动声明交换机、队列和绑定——生产者声明的 Exchange、消费者声明的 Queue 及其 Binding都会被自动在 RabbitMQ 中创建。SlimMessageBus RabbitMQ 传输中多种消息类型共用同一交换机的示意图队列与路由键支持通配符匹配Topic 类型交换机下SlimMessageBus 完整支持 RabbitMQ 的通配符路由键模式匹配规则示例*恰好一个片段regions.na.cities.*匹配regions.na.cities.toronto#零个或多个片段audit.events.#匹配audit.events.orders.placed#单独使用匹配所有路由键全量订阅mbb.ConsumeRegionEvent(x x .Queue(na-cities-queue) .ExchangeBinding(regions, routingKey: regions.na.cities.*)); 性能优化内部的路由键匹配器优先使用精确匹配仅在无精确命中时才做通配符模式匹配并对模式做了缓存。详见 src/SlimMessageBus.Host.RabbitMQ/Services/RoutingKeyMatcherService.cs。若消息的路由键未匹配到任何消费者可通过MessageUnrecognizedRoutingKeyHandler自定义行为默认 Ack 丢弃也可改为 Nack 送入死信队列、或 Requeue 稍后重试适合滚动发布场景。死信队列DLX失败消息不再丢失消费者处理失败时SlimMessageBus 默认会向 RabbitMQ 发送Nack消息将被路由到队列上配置的死信交换机或丢弃。推荐做法是为消费队列配置 DLXmbb.ConsumePingMessage(x x .Queue(subscriber, autoDelete: false) .ExchangeBinding(ping) // 队列将引用死信交换机指定 exchangeType 后 DLX 也会被自动创建 .DeadLetterExchange(subscriber-dlq, exchangeType: ExchangeType.Direct) .WithConsumerPingConsumer());也可以在总线级别为所有死信交换机设置默认值如统一使用 Direct 类型mbb.WithProviderRabbitMQ(cfg cfg.UseDeadLetterExchangeDefaults(durable: false, autoDelete: false, exchangeType: ExchangeType.Direct, routingKey: string.Empty));这样失败消息会进入subscriber-dlq对应的队列方便后续排查与补偿处理。重试与错误处理三种确认模式 自定义错误处理器确认模式决定至少一次还是至多一次SlimMessageBus 提供三种 Ack 确认模式默认是最安全的ConfirmAfterMessageProcessingWhenNoManualConfirmMade模式行为投递保证ConfirmAfterMessageProcessingWhenNoManualConfirmMade默认处理成功 Ack出错 Nack支持手动干预至少一次at-least-onceAckAutomaticByRabbit由 RabbitMQ 协议层自动 Ack至多一次at-most-onceAckMessageBeforeProcessing处理前立即 Ack至多一次吞吐量略低定义见 src/SlimMessageBus.Host.RabbitMQ/Config/RabbitMqMessageAcknowledgementMode.cs自定义错误处理器实现重试 N 次后进入死信实现IRabbitMqConsumerErrorHandlerT接口即可完全掌控失败消息的命运——例如瞬时故障则重入队列Requeue持久故障则 Nack 进 DLQpublic class CustomRabbitMqConsumerErrorHandlerT : IRabbitMqConsumerErrorHandlerT { public TaskProcessResult OnHandleError(T message, IConsumerContext ctx, Exception exception, int attempts) { if (exception is TransientException) return Task.FromResultProcessResult(RabbitMqProcessResult.Requeue); // 重试 return Task.FromResultProcessResult(ProcessResult.Failure); // 送死信 } } // 注册到 DI对任意消息类型生效 services.AddTransient(typeof(IRabbitMqConsumerErrorHandler), typeof(CustomRabbitMqConsumerErrorHandler)); 源码src/SlimMessageBus.Host.RabbitMQ/Consumers/IRabbitMqConsumerErrorHandler.cs。消费者内部还能通过ConsumerContext手动调用Ack()/Nack()精确控制单条消息。生产环境可靠性发布确认与断线自动恢复Publisher Confirms可选调用cfg.UsePublisherConfirms()开启发布确认Broker 拒绝NACK消息时会抛异常适合金融、订单等关键场景默认关闭以保吞吐也支持按生产者单独开启/退出。断线自动恢复RabbitMqChannelManager会无限次后台重连默认每 5 秒连接恢复后自动重建通道、重建拓扑并让所有消费者重新注册重启 RabbitMQ 容器也无需人工干预。消费者并发通过cfg.ConnectionFactory.ConsumerDispatchConcurrency调高单实例并发需要保证顺序时保持为 1。总结需求SlimMessageBus 方案自动创建 Exchange/Queue/Binding生产者/消费者声明时自动拓扑供给通配符路由键*/#模式 内置路由键匹配缓存失败消息兜底.DeadLetterExchange(...)一行配置 DLX重试逻辑自定义IRabbitMqConsumerErrorHandlerT Requeue投递保证默认 at-least-once可切换 at-most-once断线恢复自动重连 消费者自动重新注册延伸阅读docs/provider_rabbitmq.md 包含请求/响应、发布确认超时、多消费者同队列等完整内容更多示例可参考 src/Tests/SlimMessageBus.Host.RabbitMQ.Test/ 下的集成测试。【免费下载链接】SlimMessageBusLightweight message bus interface for .NET (pub/sub and request-response) with transport plugins for popular message brokers.项目地址: https://gitcode.com/gh_mirrors/sl/SlimMessageBus创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考