Azure Service Bus批量消息无法全部完成问题求助

问题场景 #

同步Function App从Azure Service Bus拉取500条批量消息，添加Service Bus元数据后将消息拼接持久化到Blob存储，再尝试完成所有消息。核心代码如下：

messages = []

with ServiceBusClient.from_connection_string(conn_str=SBconnstr) as servicebus_client:
    receiver = servicebus_client.get_subscription_receiver(topic_name=topic_name, subscription_name=subscription_name)

    with receiver:
        received_msgs = receiver.receive_messages(max_message_count=500)
        for msg in received_msgs:
            msg_json = json.loads(str(msg))
            msg_json['enqueued_timestamp'] = msg.enqueued_time_utc.isoformat() 
            msg_json['message_id'] = msg.message_id
            messages.append(msg_json)

        if messages:
            
            timestamp = datetime.utcnow()

            blob_service_client = BlobServiceClient.from_connection_string(STGconnstr) 
            blob_client = blob_service_client.get_blob_client(container=[redacted], blob=[redacted])

            blob_client.upload_blob(gzip.compress(json.dumps(messages).encode('utf-8')))
                    
            for msg in received_msgs:
                receiver.complete_message(msg)

异常现象 #

• Service Bus指标显示，每次拉取的500条消息中仅50-150条成功完成，无任何错误或警告，该现象始于1月22日UTC晚8:18
• 在Databricks运行完全相同的代码，消息完成率恢复至90%-100%

已排查动作 #

• 查看Service Bus诊断日志，未发现有效报错信息
• 确认Function App和Databricks使用最新版Service Bus SDK，尝试过2023年初版本及启用uamqp_transport=True，问题依旧

排查思路 #

1. 检查Function App的执行超时限制 #

Azure Function有默认执行超时时间（消费计划默认5分钟，高级/专用计划最长可设至60分钟）。如果批量处理+Blob上传的耗时接近或超过超时阈值，Function进程可能被强制终止，导致后续complete_message调用未执行完：

• 对比Databricks的执行时长和Function App的超时配置，确认是否存在超时截断情况
• 在Function代码中添加关键节点日志（比如开始处理消息、完成Blob上传、开始调用complete、每完成N条消息的计数），跟踪进程是否完整执行

2. 排查Service Bus会话/锁过期问题 #

• 检查Service Bus订阅的锁持续时间配置：默认60秒，如果从拉取消息到调用complete_message的总耗时超过锁时长，消息会自动解锁并重新进入队列，此时再调用complete_message会无效（SDK可能不会抛出明显错误）
• 对比Databricks和Function App的处理耗时，确认Function端是否因资源限制（如消费计划的CPU/内存配额）导致处理变慢，触发锁过期
• 尝试临时延长锁持续时间（比如设为5分钟），观察完成率是否提升

3. 排查Function App的资源限制与进程稳定性 #

• 消费计划的Function可能因资源节流、冷启动或进程回收导致代码执行中断：查看Function App的平台日志（如App Service日志中的kudu日志、进程回收记录），确认是否存在异常重启
• 切换到高级计划临时测试，排除消费计划的资源限制影响
• 检查Function App的并发配置：如果存在多实例并发拉取，确认是否存在消息重复处理或锁冲突

4. 捕获complete_message的隐性错误 #

当前代码未对complete_message调用做异常捕获，可能部分消息完成时出现隐性错误（如网络波动、锁已过期）但未被记录：

• 在complete_message调用处添加try-except块，捕获所有异常并记录详细日志（包括message_id、异常信息）
• 统计成功完成和失败的消息数量，对比Service Bus指标，确认是否是部分消息完成失败但未上报

5. 排查Azure平台区域级问题 #

既然问题始于特定时间点，可能和Azure Service Bus或Function App所在区域的平台事件有关：

• 查看对应区域的Azure状态页面，确认1月22日UTC晚8:18左右是否有Service Bus或App Service的服务中断或性能降级事件
• 尝试将Function App临时切换到其他区域（或Service Bus跨区域复制测试），观察问题是否消失

6. 检查消息本身的特殊性 #

• 统计未完成消息的特征：是否是特定大小、特定内容的消息？比如超大消息导致处理耗时过长，或消息内容解析异常导致后续流程中断
• 在代码中添加每条消息的处理耗时日志，定位是否有个别消息拖慢整体流程

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