基于Bluez的Python GATT服务器无法处理客户端断开连接请求

我之前在树莓派3B上用Bluez 5.49开发GATT服务时，也碰到过一模一样的客户端断开失败问题——客户端发断开请求没反应，只能硬关服务器或者客户端蓝牙。结合你提供的代码来看，问题大概率出在没有正确处理连接断开的DBus信号，以及广告/服务资源没有在断开时及时清理。

问题根源分析 #

1. Bluez的GATT框架依赖DBus信号传递连接状态变化，你的代码里没有监听org.bluez.Device1的Connected属性变化，导致断开事件发生时服务器没有任何响应
2. 广告模块的Release方法虽然实现了，但默认情况下只有当Bluez需要回收广告资源时才会触发，不会主动在断开连接时调用
3. 主代码里同时启动广告和GATT服务，但两者的生命周期没有关联，断开后资源没有重置，导致后续的断开请求被阻塞

具体修改步骤 #

1. 补全并完善广告模块代码 #

你提供的TestAdvertisement代码没写完，先补全并添加必要的广告参数：

class TestAdvertisement(Advertisement):
    def __init__(self, bus, index):
        Advertisement.__init__(self, bus, index, 'peripheral')
        self.add_service_uuid('b5fd')
        self.add_service_uuid('dfe8')
        self.add_service_uuid('8ce1')
        # 开启发射功率展示，可选
        self.include_tx_power = True

2. 修改广告模块的启动函数，返回关键引用 #

让advertising_main返回广告管理器和广告实例，方便后续断开时重新注册广告：

def advertising_main(mainloop, bus, adapter_name):
    adapter = adapters.find_adapter(bus, adapter_name)

    if not adapter:
        print('LEAdvertisingManager1 interface not found')
        return None, None

    # 确保蓝牙适配器处于开启状态
    adapter_props = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter), DBUS_PROP_IFACE)
    adapter_props.Set("org.bluez.Adapter1", "Powered", dbus.Boolean(True))

    adv_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter), LE_ADVERTISING_MANAGER_IFACE)
    adv = TestAdvertisement(bus, 0)

    try:
        adv_manager.RegisterAdvertisement(
            adv.get_path(), {},
            reply_handler=lambda: print('Advertisement registered successfully'),
            error_handler=lambda err: print(f'Failed to register advertisement: {str(err)}')
        )
    except dbus.exceptions.DBusException as e:
        print(f'Error registering advertisement: {str(e)}')
        return None, None

    print('Advertising service started')
    return adv_manager, adv

3. 在GATT服务器中添加连接状态监听 #

修改gatt_server.py，监听设备的连接状态变化，当客户端断开时清理资源并重新启动广告：

# 在gatt_server.py顶部添加必要的常量
DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
DEVICE_IFACE = 'org.bluez.Device1'

# 修改GattService类，添加信号接收器
class GattService(dbus.service.Object):
    def __init__(self, bus, path, adv_manager=None, adv=None):
        dbus.service.Object.__init__(self, bus, path)
        self.bus = bus
        # 保存广告相关引用，用于断开后重启广告
        self.adv_manager = adv_manager
        self.adv = adv

        # 监听设备连接状态变化
        self.bus.add_signal_receiver(
            self.on_device_connection_changed,
            signal_name='PropertiesChanged',
            dbus_interface=DBUS_PROP_IFACE,
            arg0=DEVICE_IFACE,
            path_keyword='device_path'
        )

    def on_device_connection_changed(self, interface, changed_props, invalidated_props, device_path):
        if interface != DEVICE_IFACE:
            return
        
        if 'Connected' in changed_props:
            is_connected = changed_props['Connected']
            if not is_connected:
                print(f"Client {device_path} disconnected successfully")
                # 重新注册广告，让设备可以被再次发现
                if self.adv_manager and self.adv:
                    try:
                        # 先注销旧广告，再重新注册
                        self.adv_manager.UnregisterAdvertisement(self.adv.get_path())
                        self.adv_manager.RegisterAdvertisement(
                            self.adv.get_path(), {},
                            reply_handler=lambda: print('Advertisement restarted'),
                            error_handler=lambda err: print(f'Failed to restart advertisement: {str(err)}')
                        )
                    except dbus.exceptions.DBusException as e:
                        print(f'Error restarting advertisement: {str(e)}')

4. 修改主函数，传递广告引用给GATT服务器 #

更新主代码，把广告管理器和实例传给GATT服务，让它能在断开时操作广告：

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('-a', '--adapter-name', type=str, help='Adapter name', default='')
    args = parser.parse_args()
    adapter_name = args.adapter_name

    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    mainloop = GObject.MainLoop()

    # 启动广告服务，获取引用
    adv_manager, adv = advertising.advertising_main(mainloop, bus, adapter_name)
    if not adv_manager or not adv:
        print('Failed to start advertising service')
        return

    # 启动GATT服务器，传递广告引用
    gatt_server.gatt_server_main(mainloop, bus, adapter_name, adv_manager, adv)

    mainloop.run()

额外注意事项 #

• 确保你的GATT特征/服务的读写方法没有阻塞操作，如果有长时间运行的任务，要放到异步线程里，避免阻塞DBus事件循环
• 如果升级Bluez到5.50+版本，这个问题可能会被官方修复，但上面的修改在5.49版本下是有效的
• 测试时可以用bluetoothctl命令行工具查看连接状态：bluetoothctl devices、bluetoothctl info <设备MAC>

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