如何在Python SDK中让Azure Service Bus客户端与发送者实例存活超10分钟？

方案1：调整内置心跳间隔（推荐） #

Azure Service Bus Python SDK的ServiceBusClient构造函数提供了keep_alive_interval参数，该参数控制SDK主动发送AMQP心跳帧的间隔，默认值为600秒（10分钟）。你可以将其设置为小于10分钟的数值（比如300秒，即5分钟），让SDK自动定期发送心跳，刷新AMQP连接的空闲超时计时器，避免连接断开。

示例代码（异步场景）：

from azure.servicebus.aio import ServiceBusClient

# 初始化客户端时设置keep_alive_interval为300秒
servicebus_client = ServiceBusClient.from_connection_string(
    conn_str="YOUR_CONNECTION_STRING",
    keep_alive_interval=300
)

# 后续创建sender的逻辑不变
async with servicebus_client:
    sender = servicebus_client.get_queue_sender(queue_name="YOUR_QUEUE")
    # 保存sender实例供后续事件触发时使用

这个方案无需额外代码，由SDK自动维护连接活跃状态，且心跳帧是AMQP协议的轻量级控制帧，不会占用过多资源。

方案2：手动触发轻量级AMQP调用（替代方案） #

如果不想调整心跳间隔，你可以定期调用队列的轻量级查询接口（比如获取队列运行时信息），这类操作会产生AMQP请求，从而刷新空闲超时计时器，且不会发送业务消息。

示例代码（异步场景）：

import asyncio
from azure.servicebus.aio import ServiceBusClient, ServiceBusSender

# 全局保存sender实例
sender: ServiceBusSender = None

async def keep_connection_alive():
    global sender
    while True:
        if sender and not sender.closed:
            # 调用轻量级接口获取队列运行时信息
            await sender.get_queue_runtime_info()
        # 每隔9分钟执行一次（小于10分钟的空闲超时）
        await asyncio.sleep(540)

async def init_servicebus():
    global sender
    servicebus_client = ServiceBusClient.from_connection_string("YOUR_CONNECTION_STRING")
    sender = servicebus_client.get_queue_sender(queue_name="YOUR_QUEUE")
    # 启动保活任务
    asyncio.create_task(keep_connection_alive())

# 初始化时启动保活任务
asyncio.run(init_servicebus())

注意：get_queue_runtime_info()是SDK提供的官方接口，不属于内部私有方法，稳定性有保障，且操作轻量，不会产生业务消息开销。

避免直接操作底层AMQP连接 #

不建议尝试获取底层AMQP连接对象（比如_amqp_client相关属性），这类属于SDK内部实现，版本更新时可能会发生变化，导致代码兼容性问题。

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