Python中多循环函数并行执行的最佳实践咨询

我有三个需要并行执行直至程序退出的循环，分别是WebSocket连接、串口读取、基于串口和WebSocket更新UI的Tkinter循环。当前用了如下异步实现方案，但觉得不够优化，想了解最佳实践：

class Window(tk.Tk):
    def __init__(self, loop,app):
        self.loop = loop
        self.app = app
        self.root = tk.Tk()

    async def show(self):
        while True:
            self.root.update()
            await asyncio.sleep(0.1)

class App:
    async def exec(self):
        self.window = Window(asyncio.get_event_loop(),self)
        await asyncio.gather(self.window.show(), self.connectWS(),self.readSerial())

    async def readSerial(self):
        self.serial = serial.serial_for_url('/dev/cu.usbserial-1430', baudrate=9600, timeout=5)
        self.serial.isOpen()
        self.serial.flushInput() #flush input buffer, discarding all its contents
        self.serial.flushOutput()
        while True:
            response = self.serial.read(1)
            await asyncio.sleep(0.1)

    async def connectWS(self):
        try:
            async with websockets.connect(
                "ws://mysocket.com") as ws:
                self.ws = WebSocketHandler(str(uuid.getnode()), ws,self)
                await asyncio.gather(self.ws.start(), self.ws.send_boot_payload())
        except websockets.exceptions.ConnectionClosedError:
            print("DISCONNECTED")
            call_later(10,self.connectWS) #reconnect

asyncio.run(App().exec())

核心优化方案 #

1. 修复Tkinter循环的异步集成问题 #

当前用root.update()加固定间隔休眠的轮询方式，效率低且易引发UI卡顿。正确做法是让Tkinter原生主循环与asyncio事件循环协同，避免手动轮询：

• 推荐用第三方库简化：使用async-tkinter-loop直接实现异步UI循环

import asyncio
from async_tkinter_loop import async_mainloop
import tkinter as tk

class Window(tk.Tk):
    def __init__(self, app):
        super().__init__()
        self.app = app
        # 初始化UI组件

# 在App中启动UI循环
async def run_ui(window):
    await async_mainloop(window)

• 原生实现方案：用Tkinter的after方法触发asyncio事件循环迭代，避免阻塞

class Window(tk.Tk):
    def __init__(self, loop, app):
        super().__init__()
        self.loop = loop
        self.app = app
        self._sync_loop()

    def _sync_loop(self):
        self.loop.call_soon(self.loop.stop)
        self.loop.run_forever()
        self.after(10, self._sync_loop)

另外注意：你的Window类继承自tk.Tk却又初始化了self.root = tk.Tk()，会创建两个Tk实例，直接调用super().__init__()即可。

2. 串口读取异步化，避免阻塞事件循环 #

当前的serial.read()是同步阻塞调用，会卡住整个asyncio事件循环。有两种优化方案：

• 方案一：用异步串口库aioserial（推荐）

from aioserial import AioSerial

async def readSerial(self):
    self.serial = AioSerial(port='/dev/cu.usbserial-1430', baudrate=9600, timeout=5)
    self.serial.flushInput()
    self.serial.flushOutput()
    while True:
        # 异步读取，不阻塞事件循环
        response = await self.serial.read_async(1)
        if response:
            # 处理串口数据，比如更新UI
            await self.window.update_serial_data(response)

• 方案二：线程+队列适配同步串口
  如果不想引入第三方库，把串口读取放到独立线程，用队列传递数据给asyncio：

import queue
import threading

async def readSerial(self):
    self.serial_queue = queue.Queue()
    # 启动后台线程处理串口读取
    threading.Thread(target=self._serial_thread, daemon=True).start()
    while True:
        # 异步从队列取数据
        response = await asyncio.get_event_loop().run_in_executor(None, self.serial_queue.get, True)
        # 处理数据

def _serial_thread(self):
    self.serial = serial.serial_for_url('/dev/cu.usbserial-1430', baudrate=9600, timeout=5)
    self.serial.flushInput()
    self.serial.flushOutput()
    while True:
        response = self.serial.read(1)
        if response:
            self.serial_queue.put(response)

3. 修复WebSocket重连逻辑 #

当前用call_later不是asyncio原生API，易引发线程问题。改用asyncio的休眠+循环实现可靠重连：

async def connectWS(self):
    while True:
        try:
            async with websockets.connect("ws://mysocket.com") as ws:
                self.ws = WebSocketHandler(str(uuid.getnode()), ws, self)
                await asyncio.gather(self.ws.start(), self.ws.send_boot_payload())
        except websockets.exceptions.ConnectionClosedError:
            print("DISCONNECTED")
        except Exception as e:
            print(f"WebSocket连接出错: {e}")
        # 等待10秒后重试
        await asyncio.sleep(10)

4. 统一事件循环管理 #

所有任务通过asyncio.create_task或asyncio.gather管理，避免手动操作事件循环实例，确保任务并行执行且可被正确取消。

完整优化示例代码 #

import asyncio
import tkinter as tk
import websockets
import uuid
from aioserial import AioSerial

class Window(tk.Tk):
    def __init__(self, app):
        super().__init__()
        self.app = app
        self.title("异步监控工具")
        # UI组件初始化
        self.serial_label = tk.Label(self, text="串口数据: 无")
        self.serial_label.pack(pady=10)
        self.ws_label = tk.Label(self, text="WebSocket状态: 未连接")
        self.ws_label.pack(pady=10)

    async def update_serial(self, data):
        # 确保UI更新在Tkinter主线程执行
        await asyncio.get_event_loop().run_in_executor(None,
            lambda: self.serial_label.config(text=f"串口数据: {data.hex()}"))

    async def update_ws_status(self, status):
        await asyncio.get_event_loop().run_in_executor(None,
            lambda: self.ws_label.config(text=f"WebSocket状态: {status}"))

class WebSocketHandler:
    def __init__(self, client_id, ws, app):
        self.client_id = client_id
        self.ws = ws
        self.app = app

    async def start(self):
        await self.app.window.update_ws_status("已连接")
        try:
            async for msg in self.ws:
                print(f"收到WebSocket消息: {msg}")
                await self.app.window.update_serial(msg.encode())
        finally:
            await self.app.window.update_ws_status("已断开")

    async def send_boot_payload(self):
        await self.ws.send(f"设备上线: {self.client_id}")

class App:
    def __init__(self):
        self.window = Window(self)

    async def exec(self):
        # 启动UI主循环任务
        ui_task = asyncio.create_task(asyncio.get_event_loop().run_in_executor(None, self.window.mainloop))
        await asyncio.gather(ui_task, self.connectWS(), self.readSerial())

    async def readSerial(self):
        self.serial = AioSerial(port='/dev/cu.usbserial-1430', baudrate=9600, timeout=5)
        self.serial.flushInput()
        self.serial.flushOutput()
        while True:
            response = await self.serial.read_async(1)
            if response:
                await self.window.update_serial(response)

    async def connectWS(self):
        while True:
            try:
                async with websockets.connect("ws://mysocket.com") as ws:
                    self.ws = WebSocketHandler(str(uuid.getnode()), ws, self)
                    await asyncio.gather(self.ws.start(), self.ws.send_boot_payload())
            except websockets.exceptions.ConnectionClosedError:
                print("WebSocket连接断开")
            except Exception as e:
                print(f"WebSocket连接失败: {e}")
            await asyncio.sleep(10)

asyncio.run(App().exec())

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