Python3.5+OSX10.13.2下多进程蓝牙EEG采集连接阻塞问题求助

你在MacOS 10.13.2上用Python3.5拆分蓝牙采集和Pygame GUI进程时，卡在socket.connect()连错误信息都没输出的情况，我之前遇到过类似的，大概率和MacOS的蓝牙权限、多进程环境下的模块兼容性有关，下面给你拆解问题和解决方案：

可能的问题根源 #

• MacOS蓝牙权限的子进程限制：MacOS对进程的蓝牙访问权限管控很严，子进程可能没法继承主进程的蓝牙访问权限，导致系统的蓝牙守护进程直接忽略了子进程的连接请求，让connect()一直阻塞，甚至连异常都抛不出来。
• PyBluez的多进程兼容性问题：你用的bluetooth模块应该是PyBluez，对应Python3.5的版本比较老，在MacOS的多进程（尤其是默认的spawn启动方式）环境下，子进程里的蓝牙Socket初始化可能不完整，直接导致连接操作无响应。
• 进程启动逻辑的小坑：你的代码里主进程启动两个进程后立刻join()，虽然理论上会等待process1完成，但如果子进程的连接操作触发了系统权限弹窗却没被处理，也会导致连接一直挂起。

分步解决方法 #

1. 先在主进程单独测试蓝牙连接，排除设备本身问题 #

先跑一段极简的连接代码，确认你的EEG设备能正常和Python通信：

import bluetooth
socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
try:
    socket.connect(('B0:B4:48:F6:38:A1', 1))
    print("connected to device!!")
    socket.close()
except Exception as e:
    print(f"failed to connect: {str(e)}")

如果这段代码也卡，先查设备是否配对、RFCOMM通道是否正确（有些设备通道不是1，用bluetooth.find_service(address='B0:B4:48:F6:38:A1')能查到正确通道号）；如果这段能正常连接，那问题就出在多进程环境上。

2. 给Socket连接加超时，捕获异常信息 #

在子进程的connect()函数里加超时，至少能让你看到连接失败的具体原因，不会一直卡着：

def connect():
    socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
    socket.settimeout(10)  # 设置10秒超时
    try:
        socket.connect(('B0:B4:48:F6:38:A1', 1))
        print("connected to device!!")
        # 这里可以添加数据读取循环
        while True:
            data = socket.recv(1024)
            if not data:
                break
            print(f"received data: {data}")
    except Exception as e:
        print(f"failed to connect/read: {str(e)}")
    finally:
        socket.close()

3. 规范多进程启动逻辑，添加进程间通信 #

MacOS的Python3.5+默认用spawn方式启动进程，这种方式下必须把主进程代码放在if __name__ == '__main__':块里，避免重复创建进程。另外，你需要用Queue实现采集进程和GUI进程的数据传递：

import multiprocessing
import time
import bluetooth
import pygame

def bluetooth_collector(queue):
    socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
    socket.settimeout(10)
    try:
        socket.connect(('B0:B4:48:F6:38:A1', 1))
        print("connected to EEG device!!")
        while True:
            # 根据设备数据包格式调整接收字节数
            data = socket.recv(32)
            if data:
                queue.put(data)
    except Exception as e:
        print(f"Collector error: {str(e)}")
    finally:
        socket.close()

def pygame_gui(queue):
    pygame.init()
    screen = pygame.display.set_mode((800, 600))
    clock = pygame.time.Clock()
    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
        # 从队列读取采集到的数据
        while not queue.empty():
            data = queue.get()
            # 这里可以添加数据处理和GUI更新逻辑
            print(f"GUI received: {data}")
        screen.fill((255,255,255))
        pygame.display.flip()
        clock.tick(25)  # 固定25Hz更新
    pygame.quit()

if __name__ == '__main__':
    data_queue = multiprocessing.Queue()
    collector_process = multiprocessing.Process(target=bluetooth_collector, args=(data_queue,))
    gui_process = multiprocessing.Process(target=pygame_gui, args=(data_queue,))
    
    collector_process.start()
    gui_process.start()
    
    # 先等待GUI进程结束，再终止采集进程
    gui_process.join()
    collector_process.terminate()
    collector_process.join()
    
    print("All processes finished")

4. 检查MacOS蓝牙权限 #

打开「系统偏好设置」→「安全与隐私」→「隐私」→「蓝牙」，确保你的Python解释器（或者你用的IDE）被允许访问蓝牙。如果没找到对应的程序，重启Python/IDE后再试。

替代方案：用线程代替进程 #

蓝牙数据采集属于IO密集型操作（等待数据包），Python的GIL不会成为瓶颈（IO操作时会自动释放GIL），所以用线程实现会更简单，还能避免多进程的权限和兼容性问题：

import threading
import time
import bluetooth
import pygame

data_queue = []
lock = threading.Lock()

def bluetooth_collector():
    socket = bluetooth.BluetoothSocket(bluetooth.RFCOMM)
    socket.settimeout(10)
    try:
        socket.connect(('B0:B4:48:F6:38:A1', 1))
        print("connected to EEG device!!")
        while True:
            data = socket.recv(32)
            if data:
                with lock:
                    data_queue.append(data)
    except Exception as e:
        print(f"Collector error: {str(e)}")
    finally:
        socket.close()

def main():
    pygame.init()
    screen = pygame.display.set_mode((800, 600))
    clock = pygame.time.Clock()
    running = True
    
    # 启动采集线程，设置为守护线程，随主进程结束
    collector_thread = threading.Thread(target=bluetooth_collector, daemon=True)
    collector_thread.start()
    
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
        # 线程安全地读取数据
        with lock:
            while data_queue:
                data = data_queue.pop(0)
                print(f"GUI received: {data}")
        screen.fill((255,255,255))
        pygame.display.flip()
        clock.tick(25)
    pygame.quit()

if __name__ == '__main__':
    main()

总结 #

先排查基础连接问题，再给Socket加超时定位异常，规范多进程启动逻辑并检查权限。如果多进程方案还是有问题，切换到线程方案完全能满足你的需求——毕竟蓝牙采集是IO密集型任务，线程的性能和复杂度都更合适。

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