asyncio+curses多协程输出重叠重复问题求助

你的代码基于asyncio和curses实现字符'Q'的界面显示，当worker协程通过ThreadPoolExecutor执行getch_nonblock处理非阻塞输入时，界面偶尔出现字符重叠、重复的现象；移除display_queue或不使用key_task时问题消失。

问题根源 #

curses库不是线程安全的。display协程在asyncio主事件循环线程中操作stdscr，而worker中的getch_nonblock在ThreadPool的子线程中也操作同一个stdscr实例。多线程并发调用curses的终端IO操作时，没有同步机制，会导致终端输出的指令被交错执行，最终出现显示混乱。

解决方案 #

方案1：用线程锁保护所有curses操作 #

通过全局线程锁，确保同一时间只有一个线程能访问curses对象，避免并发冲突。

1. 创建全局锁：

import threading
curses_lock = threading.Lock()

2. 修改display协程中的curses操作：

async def display(stdscr):
    curses.curs_set(0)
    curses.start_color()
    curses.use_default_colors()
    curses.init_pair(1, curses.COLOR_RED, -1)
    curses.init_pair(2, curses.COLOR_GREEN, -1)

    last = time.monotonic()
    while True:
        try:
            now = time.monotonic()
            to_wait = last + FRAME_INTERVAL - now
            if to_wait > 0:
                await asyncio.sleep(to_wait)
            
            # 加锁保护curses操作
            with curses_lock:
                stdscr.clear()
                stdscr.addstr(0, 0, 'Q')
                stdscr.refresh()

            last = time.monotonic()
        except Exception as e:
            pass

3. 修改getch_nonblock函数：

def getch_nonblock(stdscr):
    with curses_lock:
        return stdscr.getch()

方案2：统一所有curses操作到主事件循环线程 #

放弃线程池，直接在asyncio的主线程中处理非阻塞输入，彻底避免多线程操作curses的问题。

修改worker协程：

async def worker(stdscr):
    stdscr.nodelay(True)
    key = ''
    while True:
        try:
            # nodelay(True)下，无输入时getch()立即返回-1
            key_0 = stdscr.getch()
            if key_0 >= 0:
                key = key_0
            # 让出事件循环，避免占用CPU
            await asyncio.sleep(0.001)
        except Exception as e:
            pass

这种方案更简洁，所有curses操作都在同一个线程执行，完全规避了线程安全问题，同时性能开销也更低。

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