多Python线程/进程控制台输出换行混乱问题解决方案咨询

嘿，这个问题我太熟了——多线程或者多进程同时往命令行输出的时候，因为打印操作不是原子性的，很容易把两条内容挤在同一行，看着特别糟心。别担心，有几个靠谱的办法能彻底解决这个问题！

方法一：用锁实现独占打印 #

不管是线程还是进程，核心思路都是让打印操作变成“独占”的——同一时间只允许一个线程/进程往命令行写内容，这样就不会出现内容拼接的情况了。

线程场景（用threading.Lock） #

import threading

# 创建一个全局的打印锁
print_lock = threading.Lock()

def safe_print(message):
    # 用with语句自动获取和释放锁，避免手动操作出错
    with print_lock:
        print(message)

# 线程里调用safe_print代替直接print
def worker_thread(thread_id):
    for i in range(3):
        safe_print(f"Time{i}:Output_thread_{thread_id}")

# 启动多个线程
threads = [threading.Thread(target=worker_thread, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

进程场景（用multiprocessing.Lock） #

进程之间的锁需要通过参数传递，不能直接用全局变量：

import multiprocessing

def safe_print(message, print_lock):
    with print_lock:
        print(message)

def worker_process(proc_id, print_lock):
    for i in range(3):
        safe_print(f"Time{i}:Output_process_{proc_id}", print_lock)

if __name__ == "__main__":
    # 创建进程锁
    print_lock = multiprocessing.Lock()
    # 启动多个进程，把锁作为参数传入
    processes = [multiprocessing.Process(target=worker_process, args=(i, print_lock)) for i in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

方法二：用Python的logging模块（推荐） #

Python自带的logging模块本身就是线程安全的，针对多进程场景也有成熟的解决方案，完全不用自己手动处理锁的问题，省心又靠谱。

线程场景直接用 #

import logging
import threading

# 配置日志格式，这里直接输出原始消息
logging.basicConfig(level=logging.INFO, format="%(message)s")

def worker_thread(thread_id):
    for i in range(3):
        logging.info(f"Time{i}:Output_thread_{thread_id}")

threads = [threading.Thread(target=worker_thread, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

进程场景用QueueHandler #

多进程下直接用logging可能还是会有问题，推荐用QueueHandler把日志消息传到主进程统一打印：

import logging
import multiprocessing
from logging.handlers import QueueHandler, QueueListener
import queue

def worker_process(proc_id, log_queue):
    # 给每个进程配置QueueHandler
    logger = logging.getLogger()
    logger.addHandler(QueueHandler(log_queue))
    logger.setLevel(logging.INFO)
    
    for i in range(3):
        logger.info(f"Time{i}:Output_process_{proc_id}")

if __name__ == "__main__":
    # 创建日志队列和监听器
    log_queue = queue.Queue()
    console_handler = logging.StreamHandler()
    listener = QueueListener(log_queue, console_handler)
    listener.start()

    # 启动进程，把队列传入
    processes = [multiprocessing.Process(target=worker_process, args=(i, log_queue)) for i in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    listener.stop()

小补充：避免拆分打印 #

如果只是线程场景，还有个简单的小技巧：不要把一个完整的输出拆成多次print调用。比如不要写：

print("Time1:", "Output_process_1")

而是把内容先拼接成一个完整字符串再打印：

print(f"Time1:Output_process_1")

这样能减少被打断的概率，但这个方法不如锁或者logging可靠，只能作为辅助手段。

总的来说，用锁或者logging模块是解决这个问题的最优方案，根据你的线程/进程场景选对应的方法就行，保证每条输出都能单独占一行！

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