SlimMessageBus错误处理与重试机制实战如何配置IConsumerErrorHandler、死信队列与健康检查熔断器【免费下载链接】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和请求-响应两种模式并提供 RabbitMQ、Kafka、Azure Service Bus 等主流消息代理message broker的传输插件。当消息处理失败时SlimMessageBus 通过IConsumerErrorHandler错误处理器、带延迟抖动的重试工具以及健康检查驱动的熔断器circuit breaker帮助消费者优雅地处理异常消息避免消息丢失或消费风暴。一、为什么消息处理失败是常态在分布式系统中消费者Consumer处理消息时可能遇到数据库超时、依赖服务宕机、第三方 API 限流等问题。如果没有完善的消息总线错误处理策略失败的消息要么被悄悄丢弃要么无限重试拖垮队列。SlimMessageBus 的错误处理体系分三层单次失败决策—— 由IConsumerErrorHandlerT决定这条消息重试、失败还是成功重试节奏控制—— 由Retry工具类提供延迟与抖动jitter整体熔断保护—— 当健康检查变差时用熔断器暂停整个消费者二、IConsumerErrorHandler拦截消费者异常的入口核心接口定义在src/SlimMessageBus.Host/Consumer/ErrorHandling/IConsumerErrorHandler.cs当消费者或消息处理器handler抛出异常时框架会回调OnHandleError(message, consumerContext, exception, attempts)参数中带有失败次数 attempts你可以据此决定再试一次还是放弃consumerContext允许你执行传输层操作如确认acknowledge、拒绝reject、进入死信队列等接口注释明确建议如果打算重试应在方法返回之前应用延迟/抖动避免重试风暴该接口返回ProcessResult内置四种结果见src/SlimMessageBus.Host/Consumer/ErrorHandling/ProcessResult.cs结果含义典型用途ProcessResult.Retry立即重试不放回队列瞬时抖动、连接闪断ProcessResult.Failure消息放回队列交由 Broker 层重新投递ProcessResult.Success视为处理成功异常可容忍、幂等已处理ProcessResult.SuccessWithResponse成功并附带兜底响应请求-响应模式的容错降级传输插件的专属扩展各传输插件在通用接口之上提供了更贴合 Broker 能力的结果。以 RabbitMQ 为例src/SlimMessageBus.Host.RabbitMQ/Consumers/IRabbitMqConsumerErrorHandler.cs中定义了IRabbitMqConsumerErrorHandlerT基类额外提供Requeue()结果——将消息重新入队requeue语义上比Failure更精确。类似的专属错误处理器还有Kafkasrc/SlimMessageBus.Host.Kafka/Consumer/IKafkaConsumerErrorHandler.csAzure Service Bussrc/SlimMessageBus.Host.AzureServiceBus/Consumer/IServiceBusConsumerErrorHandler.csAmazon SQSsrc/SlimMessageBus.Host.AmazonSQS/Consumer/ISqsConsumerErrorHandler.cs三、带延迟与抖动的重试Retry 工具类盲目立即重试往往加重下游压力。SlimMessageBus 内置了Retry静态工具类src/SlimMessageBus.Host/Helpers/Retry.cs提供WithDelay方法支持三个关键能力固定延迟delay每次重试前等待指定时长随机抖动jitter在延迟上叠加随机量打散重试时间点防止多个消费者同时冲击下游可判定的重试条件通过shouldRetry委托按异常类型与次数决定是否继续重试在自定义错误处理器中典型做法是先判断异常是否为可重试类型 → 判断 attempts 是否超过上限 → 未超限则在Retry.WithDelay中等待后返回Retry()超限则转死信。四、死信队列让失败消息有处可去当一条消息多次重试后仍无法处理最佳实践是把它转入死信队列Dead Letter Queue, DLQ而不是丢弃或无限循环。SlimMessageBus 的死信策略依托于错误处理器中的consumerContext——它暴露了各传输层的原生操作能力RabbitMQ可拒绝消息并路由到死信交换机DLXKafka / Azure Service Bus可提交到专用的死信主题/死信子队列通用做法OnHandleError中判断attempts超过阈值后将消息写入独立 DLQ 路径并返回Success使主队列不再积压配合IConsumerErrorHandler的attempts参数你可以轻松实现重试 N 次后进入死信的经典模式既保留失败现场便于事后排查又不阻塞正常消息流。多个消息类型共用一个主题的场景参见官方文档图示SlimMessageBus消息总线多个消息类型共用一个主题的示意五、健康检查熔断器下游故障时主动暂停消费当依赖服务持续不可用时消费一条失败一条只是浪费资源。SlimMessageBus 提供了**熔断器circuit breaker**机制当熔断状态触发时消费者整体暂停消费状态恢复后自动恢复。核心设计熔断器接口IConsumerCircuitBreakersrc/SlimMessageBus.Host.CircuitBreaker/IConsumerCircuitBreaker.cs只需暴露StateCircuit.Open/Circuit.Closed与Subscribe订阅回调消费者拦截器CircuitBreakerConsumerInterceptorsrc/SlimMessageBus.Host.CircuitBreaker/Implementation/CircuitBreakerConsumerInterceptor.cs监听状态变化熔断合闸Closed时自动暂停pause消费者熔断打开时恢复消费并记录警告/信息日志通过ConsumerBuilderExtensions.AddConsumerCircuitBreakerTypeTBreaker()即可把自定义熔断器挂到消费者上与健康检查集成SlimMessageBus.Host.CircuitBreaker.HealthCheck插件内置了HealthCheckCircuitBreakersrc/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreaker.cs它监听 ASP.NET Core 健康检查端点中带标签tag的检查项检查项变为Unhealthy或按配置 Degraded→ 熔断合闸消费者暂停检查项恢复Healthy→ 熔断打开消费者自动恢复官方示例src/Samples/Sample.CircuitBreaker.HealthCheck/Program.cs展示了完整用法PauseOnUnhealthyCheck(tag)—— 对应标签的健康检查变为不健康时暂停该消费者PauseOnDegradedHealthCheck(tag)—— 更激进检查项降级Degraded即暂停这让你无需自己写轮询逻辑就能把/healthz端点的监控结果直接接入消息总线的消费开关实现下游病了就先歇着的自我保护。六、最佳实践速查场景推荐策略瞬时网络抖动ProcessResult.RetryRetry.WithDelay加抖动异常可容忍 / 已幂等处理返回ProcessResult.Success直接确认多次重试仍失败写入死信队列后返回Success主队列保持畅通依赖服务整体故障接入健康检查熔断器暂停消费者等待恢复请求-响应超时兜底返回ProcessResult.SuccessWithResponse提供降级响应提示各传输插件的错误处理器接口均位于其插件目录的Consumer/或Consumers/子目录下配置入口则在Config/子目录中可按所用 Broker 查阅对应扩展方法。通过IConsumerErrorHandler的精细决策、Retry的延迟抖动控制以及健康检查熔断器的整体保护SlimMessageBus 让 .NET 应用的消息消费具备了生产级的容错能力。【免费下载链接】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错误处理与重试机制实战：如何配置IConsumerErrorHandler、死信队列与健康检查熔断器
SlimMessageBus错误处理与重试机制实战如何配置IConsumerErrorHandler、死信队列与健康检查熔断器【免费下载链接】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和请求-响应两种模式并提供 RabbitMQ、Kafka、Azure Service Bus 等主流消息代理message broker的传输插件。当消息处理失败时SlimMessageBus 通过IConsumerErrorHandler错误处理器、带延迟抖动的重试工具以及健康检查驱动的熔断器circuit breaker帮助消费者优雅地处理异常消息避免消息丢失或消费风暴。一、为什么消息处理失败是常态在分布式系统中消费者Consumer处理消息时可能遇到数据库超时、依赖服务宕机、第三方 API 限流等问题。如果没有完善的消息总线错误处理策略失败的消息要么被悄悄丢弃要么无限重试拖垮队列。SlimMessageBus 的错误处理体系分三层单次失败决策—— 由IConsumerErrorHandlerT决定这条消息重试、失败还是成功重试节奏控制—— 由Retry工具类提供延迟与抖动jitter整体熔断保护—— 当健康检查变差时用熔断器暂停整个消费者二、IConsumerErrorHandler拦截消费者异常的入口核心接口定义在src/SlimMessageBus.Host/Consumer/ErrorHandling/IConsumerErrorHandler.cs当消费者或消息处理器handler抛出异常时框架会回调OnHandleError(message, consumerContext, exception, attempts)参数中带有失败次数 attempts你可以据此决定再试一次还是放弃consumerContext允许你执行传输层操作如确认acknowledge、拒绝reject、进入死信队列等接口注释明确建议如果打算重试应在方法返回之前应用延迟/抖动避免重试风暴该接口返回ProcessResult内置四种结果见src/SlimMessageBus.Host/Consumer/ErrorHandling/ProcessResult.cs结果含义典型用途ProcessResult.Retry立即重试不放回队列瞬时抖动、连接闪断ProcessResult.Failure消息放回队列交由 Broker 层重新投递ProcessResult.Success视为处理成功异常可容忍、幂等已处理ProcessResult.SuccessWithResponse成功并附带兜底响应请求-响应模式的容错降级传输插件的专属扩展各传输插件在通用接口之上提供了更贴合 Broker 能力的结果。以 RabbitMQ 为例src/SlimMessageBus.Host.RabbitMQ/Consumers/IRabbitMqConsumerErrorHandler.cs中定义了IRabbitMqConsumerErrorHandlerT基类额外提供Requeue()结果——将消息重新入队requeue语义上比Failure更精确。类似的专属错误处理器还有Kafkasrc/SlimMessageBus.Host.Kafka/Consumer/IKafkaConsumerErrorHandler.csAzure Service Bussrc/SlimMessageBus.Host.AzureServiceBus/Consumer/IServiceBusConsumerErrorHandler.csAmazon SQSsrc/SlimMessageBus.Host.AmazonSQS/Consumer/ISqsConsumerErrorHandler.cs三、带延迟与抖动的重试Retry 工具类盲目立即重试往往加重下游压力。SlimMessageBus 内置了Retry静态工具类src/SlimMessageBus.Host/Helpers/Retry.cs提供WithDelay方法支持三个关键能力固定延迟delay每次重试前等待指定时长随机抖动jitter在延迟上叠加随机量打散重试时间点防止多个消费者同时冲击下游可判定的重试条件通过shouldRetry委托按异常类型与次数决定是否继续重试在自定义错误处理器中典型做法是先判断异常是否为可重试类型 → 判断 attempts 是否超过上限 → 未超限则在Retry.WithDelay中等待后返回Retry()超限则转死信。四、死信队列让失败消息有处可去当一条消息多次重试后仍无法处理最佳实践是把它转入死信队列Dead Letter Queue, DLQ而不是丢弃或无限循环。SlimMessageBus 的死信策略依托于错误处理器中的consumerContext——它暴露了各传输层的原生操作能力RabbitMQ可拒绝消息并路由到死信交换机DLXKafka / Azure Service Bus可提交到专用的死信主题/死信子队列通用做法OnHandleError中判断attempts超过阈值后将消息写入独立 DLQ 路径并返回Success使主队列不再积压配合IConsumerErrorHandler的attempts参数你可以轻松实现重试 N 次后进入死信的经典模式既保留失败现场便于事后排查又不阻塞正常消息流。多个消息类型共用一个主题的场景参见官方文档图示SlimMessageBus消息总线多个消息类型共用一个主题的示意五、健康检查熔断器下游故障时主动暂停消费当依赖服务持续不可用时消费一条失败一条只是浪费资源。SlimMessageBus 提供了**熔断器circuit breaker**机制当熔断状态触发时消费者整体暂停消费状态恢复后自动恢复。核心设计熔断器接口IConsumerCircuitBreakersrc/SlimMessageBus.Host.CircuitBreaker/IConsumerCircuitBreaker.cs只需暴露StateCircuit.Open/Circuit.Closed与Subscribe订阅回调消费者拦截器CircuitBreakerConsumerInterceptorsrc/SlimMessageBus.Host.CircuitBreaker/Implementation/CircuitBreakerConsumerInterceptor.cs监听状态变化熔断合闸Closed时自动暂停pause消费者熔断打开时恢复消费并记录警告/信息日志通过ConsumerBuilderExtensions.AddConsumerCircuitBreakerTypeTBreaker()即可把自定义熔断器挂到消费者上与健康检查集成SlimMessageBus.Host.CircuitBreaker.HealthCheck插件内置了HealthCheckCircuitBreakersrc/SlimMessageBus.Host.CircuitBreaker.HealthCheck/HealthCheckCircuitBreaker.cs它监听 ASP.NET Core 健康检查端点中带标签tag的检查项检查项变为Unhealthy或按配置 Degraded→ 熔断合闸消费者暂停检查项恢复Healthy→ 熔断打开消费者自动恢复官方示例src/Samples/Sample.CircuitBreaker.HealthCheck/Program.cs展示了完整用法PauseOnUnhealthyCheck(tag)—— 对应标签的健康检查变为不健康时暂停该消费者PauseOnDegradedHealthCheck(tag)—— 更激进检查项降级Degraded即暂停这让你无需自己写轮询逻辑就能把/healthz端点的监控结果直接接入消息总线的消费开关实现下游病了就先歇着的自我保护。六、最佳实践速查场景推荐策略瞬时网络抖动ProcessResult.RetryRetry.WithDelay加抖动异常可容忍 / 已幂等处理返回ProcessResult.Success直接确认多次重试仍失败写入死信队列后返回Success主队列保持畅通依赖服务整体故障接入健康检查熔断器暂停消费者等待恢复请求-响应超时兜底返回ProcessResult.SuccessWithResponse提供降级响应提示各传输插件的错误处理器接口均位于其插件目录的Consumer/或Consumers/子目录下配置入口则在Config/子目录中可按所用 Broker 查阅对应扩展方法。通过IConsumerErrorHandler的精细决策、Retry的延迟抖动控制以及健康检查熔断器的整体保护SlimMessageBus 让 .NET 应用的消息消费具备了生产级的容错能力。【免费下载链接】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），仅供参考