Python多线程打印内容串行问题及并发打印解决方案需求

问题场景 #

你用Python threading模块实现了多线程打印功能，支持创建任意数量线程，但运行时出现打印消息被拼接在同一行的异常情况。

原代码 #

import threading
import time
import random


class Useless:
    def __init__(self, numb):
        self.numb = numb

    def printing(self):
        time.sleep(random.uniform(0.05, 0.09))
        print(f'number {self.numb} is being printed God knows what for...')

    def main_loop(self):
        for _ in range(5):
            self.printing()


obj_1 = Useless(1)
obj_2 = Useless(2)

proc_1 = threading.Thread(target=obj_1.main_loop, args=())
proc_2 = threading.Thread(target=obj_2.main_loop, args=())

proc_1.start()
proc_2.start()

异常输出示例 #

number 2 is being printed God knows what for...number 1 is being printed God knows what for...

number 1 is being printed God knows what for...
number 2 is being printed God knows what for...
number 1 is being printed God knows what for...
number 2 is being printed God knows what for...
number 1 is being printed God knows what for...
number 2 is being printed God knows what for...
number 1 is being printed God knows what for...
number 2 is being printed God knows what for...

Process finished with exit code 0

需求：保持线程并发执行的前提下，实现打印消息各行独立。

解决方案 #

核心原因 #

Python的print操作不是原子性的，多个线程同时执行print时，可能会打断彼此的输出流程，导致部分内容拼接在同一行。

方法一：线程锁（最直接高效） #

给打印操作加线程锁，确保同一时间只有一个线程能执行打印，保证输出的原子性。修改后的代码支持创建任意数量线程：

import threading
import time
import random

# 全局打印锁
print_lock = threading.Lock()

class Useless:
    def __init__(self, numb):
        self.numb = numb

    def printing(self):
        time.sleep(random.uniform(0.05, 0.09))
        # 使用with自动获取/释放锁，避免遗漏释放操作
        with print_lock:
            print(f'number {self.numb} is being printed God knows what for...')

    def main_loop(self):
        for _ in range(5):
            self.printing()

# 创建任意数量线程示例（这里创建3个）
threads = []
for thread_num in range(1, 4):
    obj = Useless(thread_num)
    t = threading.Thread(target=obj.main_loop)
    threads.append(t)
    t.start()

# 可选：等待所有线程执行完毕
for t in threads:
    t.join()

方法二：asyncio异步实现 #

如果想用异步方案，asyncio的事件循环是单线程调度，打印操作不会出现多线程竞争问题，同样能实现并发效果：

import asyncio
import random

class Useless:
    def __init__(self, numb):
        self.numb = numb

    async def printing(self):
        # 异步睡眠，不阻塞事件循环
        await asyncio.sleep(random.uniform(0.05, 0.09))
        print(f'number {self.numb} is being printed God knows what for...')

    async def main_loop(self):
        for _ in range(5):
            await self.printing()

async def main():
    # 创建任意数量异步任务
    tasks = []
    for task_num in range(1, 4):
        obj = Useless(task_num)
        tasks.append(asyncio.create_task(obj.main_loop()))
    # 等待所有任务完成
    await asyncio.gather(*tasks)

asyncio.run(main())

关于subprocess pipe的说明 #

subprocess用于创建独立子进程，进程间通信成本高，对于这个打印混乱的问题完全没必要使用，线程锁或asyncio都是更轻量高效的方案。

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