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），僅供參考 SlimMessageBus對接RabbitMQ完全指南：隊列、死信與重試錯誤處理實戰
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），僅供參考