Azure Service Bus队列循环读取报ServiceUnavailableError问题咨询

错误原因 #

你遇到的ServiceUnavailableError: Operation to settle the message has timed out错误，核心原因是频繁创建并关闭Service Bus Receiver实例，导致消息处理流程被强行中断：

• 当你在循环中每5秒新建一个Receiver、注册处理器后立刻关闭它时，很可能此时还有未处理完成的消息（比如brokeredMessage.complete()还在执行）。关闭Receiver会直接终止当前的AMQP连接会话，导致正在进行的消息确认操作（settle）超时。
• 重复创建/销毁Receiver会快速消耗AMQP连接资源，Service Bus对连接数有配额限制，频繁操作会导致连接池耗尽，进而触发服务不可用的错误。

你的代码问题 #

这种读取消息的方式完全不正确，主要问题有：

• 完全没必要在循环中重复创建Receiver：Service Bus的Receiver本身就是为长时监听设计的，一旦注册了registerMessageHandler，它会持续自动从队列拉取消息，不需要循环重建实例。
• 短时间内创建/关闭Receiver会破坏消息处理的原子性：当maxConcurrentCalls设为10时，可能同时有10个消息在处理流程中，此时关闭Receiver会直接中断这些处理中的消息确认操作，引发超时报错。

正确的实现方式 #

只需要创建一次Receiver并注册消息处理器，让它持续运行即可。如果需要停止监听，再主动优雅关闭Receiver、QueueClient和ServiceBusClient。

修正后的代码示例：

const { ServiceBusClient, ReceiveMode, delay } = require("@azure/service-bus");
const connectionString = "Endpoint=sb://some-url";
const queueName = "my-queue";

async function main() {
  const sbClient = ServiceBusClient.createFromConnectionString(connectionString);
  const queueClient = sbClient.createQueueClient(queueName);
  const receiver = queueClient.createReceiver(ReceiveMode.peekLock);

  const onMessageHandler = async (brokeredMessage) => {
    console.log(`Received message: ${brokeredMessage.body}`);
    // 给消息确认操作单独加异常捕获，避免影响后续消息处理
    try {
      await brokeredMessage.complete();
    } catch (err) {
      console.error(`Failed to complete message: ${err}`);
      // 确认失败时可以选择放弃消息，让它重新进入队列
      await brokeredMessage.abandon();
    }
  };

  const onErrorHandler = (err) => {
    console.log("Error occurred: ", err);
    // 针对可重试的错误，可以加入自动重试逻辑
    if (err.retryable) {
      console.log("Retrying message receiver...");
      // 这里可以实现延迟后重新注册处理器的逻辑
    }
  };

  try {
    // 注册处理器后，Receiver会持续监听队列
    receiver.registerMessageHandler(onMessageHandler, onErrorHandler, { 
      autoComplete: false, 
      maxConcurrentCalls: 10,
      // 可选：设置消息锁自动延长时间，避免处理超时导致锁过期
      maxAutoRenewDuration: 30000 
    });
    console.log("Receiver started, listening for messages...");
    // 让程序持续运行，监听终止信号（比如Ctrl+C）
    await new Promise((resolve) => process.on("SIGINT", resolve));
  } catch (err) {
    console.error("Main process error: ", err);
  } finally {
    // 优雅关闭所有资源
    await receiver.close();
    await queueClient.close();
    await sbClient.close();
    console.log("All Service Bus resources closed");
  }
}

main().catch((err) => {
  console.log("Error occurred: ", err);
});

额外建议 #

• 启用maxAutoRenewDuration：当消息处理时间较长时，自动延长消息的锁定时间，避免锁过期导致消息被重新投递。
• 给消息确认操作单独加异常捕获：避免单个消息的确认失败影响整个处理器的运行。
• 复用Service Bus客户端实例：ServiceBusClient、QueueClient和Receiver都是重量级对象，应该复用而不是频繁创建销毁。

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