如何在Shell中优雅实现并发任务的实时状态输出？

1. 利用ANSI终端转义序列实现固定行更新 #

给每个设备分配终端内的固定行，更新状态时通过光标移动回到该行开头，覆盖原有内容，实现单设备状态的实时刷新。

示例代码：

import concurrent.futures
import sys
from itertools import repeat

# 为每个设备分配固定行号
device_lines = {dev_id: i+1 for i, dev_id in enumerate(ids)}

def device_ctl(dev_id, args):
    def update_status(status):
        line_num = device_lines[dev_id]
        # 移动光标到指定行开头，打印状态并用空格覆盖旧内容
        sys.stdout.write(f"\033[{line_num};1H{dev_id} {status:<30}")
        sys.stdout.flush()

    update_status("Connecting...")
    # 模拟设备通信逻辑
    # ...
    update_status("Connected! Status: Idle")
    # 后续状态更新直接调用update_status即可
    # ...

with concurrent.futures.ThreadPoolExecutor(20) as executor:
    executor.map(device_ctl, ids, repeat(args))

# 最后将光标移到所有设备行下方，避免后续输出覆盖状态
sys.stdout.write(f"\033[{len(ids)+1};1H\n")
sys.stdout.flush()

2. 单线程统一处理输出（生产者-消费者模式） #

创建专门的输出线程，所有设备线程仅负责收集状态并发送到队列，由输出线程统一维护和刷新终端内容，彻底避免多线程竞争输出资源。

示例代码：

import concurrent.futures
import queue
import threading
from itertools import repeat

status_queue = queue.Queue()
device_status = {}

def output_worker():
    while True:
        try:
            dev_id, status = status_queue.get()
            if dev_id is None:  # 接收结束信号
                break
            device_status[dev_id] = status
            # 清屏后重新打印所有设备状态（适合设备数量不多的场景）
            print("\033[H\033[J", end="")
            for dev in device_status:
                print(f"{dev} {device_status[dev]}")
            status_queue.task_done()
        except Exception:
            break

# 启动输出线程
output_thread = threading.Thread(target=output_worker)
output_thread.start()

def device_ctl(dev_id, args):
    status_queue.put((dev_id, "Connecting..."))
    # 模拟设备通信逻辑
    # ...
    status_queue.put((dev_id, "Connected! Status: Idle"))
    # 后续状态更新同理
    # ...

with concurrent.futures.ThreadPoolExecutor(20) as executor:
    executor.map(device_ctl, ids, repeat(args))

# 发送结束信号并等待输出线程完成
status_queue.put((None, None))
output_thread.join()

3. 使用第三方终端库简化实现 #

借助rich这类封装好的终端UI库，无需手动处理ANSI序列，快速实现美观的实时状态展示。

示例代码（需先安装：pip install rich）：

from rich.live import Live
from rich.table import Table
import concurrent.futures
from itertools import repeat
import threading

# 初始化状态表格
status_table = Table(show_header=False)
for dev_id in ids:
    status_table.add_row(dev_id, "Connecting...")

table_lock = threading.Lock()

def device_ctl(dev_id, args):
    with table_lock:
        # 定位对应设备行并更新状态
        for row in status_table.rows:
            if row.cells[0].value == dev_id:
                row.cells[1].value = "Connected! Status: Idle"
                break
    # 后续状态更新同理
    # ...

# 启动实时展示
with Live(status_table, refresh_per_second=4):
    with concurrent.futures.ThreadPoolExecutor(20) as executor:
        executor.map(device_ctl, ids, repeat(args))

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