多线程CFD任务类打印输出混乱问题的解决咨询

一、使用threading.Lock实现输出同步 #

print本身不是线程安全的操作，多个线程同时调用时，会出现输出内容拆分混行的问题。用threading.Lock可以强制打印操作串行执行，保证每行输出完整有序。

实现步骤： #

1. 定义一个全局的锁实例，所有线程共享该锁
2. 把分散的print调用封装成统一的日志方法，在方法内通过锁控制打印过程

代码示例：

import threading

# 全局打印锁，所有线程共用
print_lock = threading.Lock()

class CFDTaskManager:
    def __init__(self, dir_path):
        self.dir_path = dir_path
        # 其他初始化逻辑（文件校验、bat脚本生成等）

    def _safe_print(self, content):
        # 用with语句自动管理锁的获取与释放，避免遗漏释放操作
        with print_lock:
            print(f"[{self.dir_path}] {content}")

    def main_loop(self):
        # 校验文件
        self._safe_print("开始校验目录文件")
        # 省略校验逻辑...
        self._safe_print("文件校验完成")

        # 生成bat脚本
        self._safe_print("生成CFD调用脚本")
        # 省略脚本生成逻辑...
        self._safe_print("脚本生成完成")

        # 启动CFD任务并监控
        self._safe_print("启动CFD计算任务")
        # 省略subprocess调用与监控逻辑...
        self._safe_print("CFD任务完成")

每次打印时，线程会先获取锁，打印完成后自动释放锁，确保同一时间只有一个线程在输出，彻底解决混行问题。

二、使用消息队列集中处理输出 #

创建一个专门负责输出的独立线程，所有工作线程不直接调用print，而是把要输出的消息放到队列中，由输出线程统一取出打印。这种方式天然保证输出有序，还能灵活扩展输出方式（比如同时写入日志文件）。

实现步骤： #

1. 创建全局消息队列
2. 启动输出线程，循环从队列中取消息并打印
3. 工作线程将输出内容放入队列，由输出线程统一处理

代码示例：

import threading
import queue

# 全局输出消息队列
output_queue = queue.Queue()

def output_worker():
    """专门处理输出的线程，daemon=True表示随主进程退出"""
    while True:
        msg = output_queue.get()
        if msg is None:  # 用None作为线程终止信号
            break
        print(msg)
        output_queue.task_done()

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

class CFDTaskManager:
    def __init__(self, dir_path):
        self.dir_path = dir_path
        # 其他初始化逻辑

    def _send_log(self, content):
        # 把消息放入队列，由输出线程处理
        log_msg = f"[{self.dir_path}] {content}"
        output_queue.put(log_msg)

    def main_loop(self):
        self._send_log("开始校验目录文件")
        # 省略校验逻辑
        self._send_log("文件校验完成")

        self._send_log("生成CFD调用脚本")
        # 省略脚本生成逻辑
        self._send_log("脚本生成完成")

        self._send_log("启动CFD计算任务")
        # 省略任务启动与监控逻辑
        self._send_log("CFD任务完成")

# 批量任务执行示例
if __name__ == "__main__":
    task_dirs = ["dir1", "dir2", "dir3"]
    threads = []
    for dir_path in task_dirs:
        manager = CFDTaskManager(dir_path)
        thread = threading.Thread(target=manager.main_loop)
        threads.append(thread)
        thread.start()

    # 等待所有任务线程完成
    for thread in threads:
        thread.join()

    # 发送终止信号，关闭输出线程
    output_queue.put(None)
    output_thread.join()

这种方案的优势在于：输出逻辑与业务逻辑解耦，后续如果需要修改输出方式（比如写入日志文件、格式化输出），只需要修改output_worker函数，不需要改动各个CFD任务的代码。

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