Azure ServiceBus Scala消费者：如何实现重试策略与死信队列处理？

我基于Azure官方教程实现了一个简单的ServiceBus消费者Scala代码：

val consumeFunc = (context: ServiceBusReceivedMessageContext) => consumeMessageOrder(context, onConsumed)
val queueName = messageType match {
  case MessageBusService.Order => config.ordersQueueName
}

new ServiceBusClientBuilder()
  .connectionString(config.connectionString)
  .processor()
  .queueName(queueName)
  .processMessage { context =>
    consumeFunc(context)
  }
  .processError(context => processErrorOrder(context, countdownLatch))
  .buildProcessorClient()

对应的错误处理方法：

def processErrorOrder(context: ServiceBusErrorContext, countdownLatch: CountDownLatch): Unit = {
  import ServiceBusFailureReason._
  context.getException match {
    case e: ServiceBusException
        if List(MESSAGING_ENTITY_DISABLED, MESSAGING_ENTITY_NOT_FOUND, UNAUTHORIZED).contains(
          e.getReason
        ) =>
      log.error(s"An unrecoverable error occurred. Stopping processing with reason ${e.getReason} ${e.getMessage}")
      countdownLatch.countDown()
    case e: ServiceBusException =>
      log.error(s"Error source ${context.getErrorSource}, reason ${e.getReason}, message: ${e}")
  }
}

目前我无法找到实现重试策略的方法，想请教如何在此基础上处理重试，以及当重试X次失败后将消息移入死信队列？

实现重试与死信处理的方案 #

1. 利用Azure Service Bus内置重试配置 #

Azure Service Bus SDK提供内置重试策略，可在构建客户端时直接配置：

import com.azure.messaging.servicebus.ServiceBusRetryOptions
import java.time.Duration

// 自定义重试规则：最多重试3次，初始间隔2秒，最大间隔10秒
val retryOptions = new ServiceBusRetryOptions()
  .setMaxRetries(3)
  .setDelay(Duration.ofSeconds(2))
  .setMaxDelay(Duration.ofSeconds(10))

val processor = new ServiceBusClientBuilder()
  .connectionString(config.connectionString)
  .processor()
  .queueName(queueName)
  .retryOptions(retryOptions) // 注入重试配置
  .processMessage { context =>
    consumeFunc(context)
  }
  .processError(context => processErrorOrder(context, countdownLatch))
  .buildProcessorClient()

说明：内置重试会自动处理网络波动、服务暂不可用等可重试异常，重试失败后触发processError回调。

2. 手动控制重试次数与死信逻辑 #

如果需要针对业务异常做精细化控制，可在消息处理方法中手动跟踪投递次数，达到阈值后移入死信队列：

步骤1：修改消息处理逻辑

def consumeMessageOrder(context: ServiceBusReceivedMessageContext, onConsumed: => Unit): Unit = {
  val message = context.getMessage()
  val maxRetryTimes = 3 // 自定义最大重试次数
  val currentDeliveryCount = message.getDeliveryCount()

  try {
    // 执行业务逻辑
    onConsumed
    // 处理成功，标记消息完成
    context.complete()
  } catch {
    case e: Exception =>
      if (currentDeliveryCount >= maxRetryTimes) {
        // 达到重试上限，移入死信队列
        log.warn(s"Message ${message.getMessageId()} failed after $maxRetryTimes attempts, moving to dead-letter queue")
        context.deadLetter()
      } else {
        // 放弃消息，触发Service Bus重新投递（投递次数+1）
        log.error(s"Message processing failed, retry count: $currentDeliveryCount, will retry", e)
        context.abandon()
      }
  }
}

步骤2：优化错误处理回调

def processErrorOrder(context: ServiceBusErrorContext, countdownLatch: CountDownLatch): Unit = {
  import ServiceBusFailureReason._
  context.getException match {
    case e: ServiceBusException
        if List(MESSAGING_ENTITY_DISABLED, MESSAGING_ENTITY_NOT_FOUND, UNAUTHORIZED).contains(
          e.getReason
        ) =>
      log.error(s"Unrecoverable error: ${e.getReason} - ${e.getMessage}, stopping processor")
      countdownLatch.countDown()
      context.getProcessorClient().stop() // 终止处理器，避免无效循环
    case e: ServiceBusException =>
      log.error(s"Processing error source: ${context.getErrorSource}, reason: ${e.getReason}", e)
  }
}

3. 核心API说明 #

• context.complete()：标记消息处理成功，从队列中移除
• context.abandon()：放弃消息，Service Bus将其重新放回队列，投递次数+1
• context.deadLetter()：将消息移入死信队列，不再自动投递
• message.getDeliveryCount()：获取当前消息的投递次数（初始为1，每重试一次递增）

内容的提问来源于stack exchange，提问作者FrancMo