使用pathos.multiprocessing实现并行进程独立打印的方法咨询

我正在使用pathos.multiprocessing模块以并行进程运行函数，每个进程传入不同的输入参数。以下是最小可复现示例：

import pathos.multiprocessing as mp
from time import sleep

def my_func(x, y):
    for i in range(x):
        print(y+i)
        sleep(.2)
    return i + y

seq = [(100, 4), (100, 5)]
processes = 2
print ("Multiprocessing...")
pool = mp.Pool(processes)
resultsObj = pool.starmap_async(my_func, seq )          
pool.close()
results = resultsObj.get()

如预期所示，两个进程的打印结果会混杂在一起，示例输出如下：

Multiprocessing...
4
5
5
6
7
6
7
8
8
9
10
9
10
11

请问是否有办法将不同进程的输出导向不同终端以监控进度？或者有没有其他方式实现“按进程独立打印”的效果？

解决方案 #

方法1：给输出添加进程/任务标识 #

直接在打印内容前加上进程ID或自定义任务编号，即使输出混杂，也能快速区分每个进程的内容。

修改后的代码示例：

import pathos.multiprocessing as mp
from time import sleep
import os

def my_func(x, y, task_id):
    proc_id = os.getpid()  # 获取当前进程ID
    for i in range(x):
        print(f"[任务{task_id}/进程{proc_id}] {y+i}")
        sleep(.2)
    return i + y

seq = [(100, 4, 1), (100, 5, 2)]  # 给每个任务添加唯一编号
processes = 2
print ("Multiprocessing...")
pool = mp.Pool(processes)
resultsObj = pool.starmap_async(my_func, seq )          
pool.close()
results = resultsObj.get()

输出示例：

Multiprocessing...
[任务1/进程1234] 4
[任务2/进程5678] 5
[任务1/进程1234] 5
[任务2/进程5678] 6

方法2：将进程输出重定向到独立文件 #

把每个进程的打印内容写入单独的日志文件，之后可以在不同终端分别监控文件的实时更新。

修改后的代码示例：

import pathos.multiprocessing as mp
from time import sleep
import sys

def my_func(x, y, log_file):
    # 重定向标准输出到指定文件
    sys.stdout = open(log_file, 'a', buffering=1)  # 行缓冲保证实时写入
    for i in range(x):
        print(y+i)
        sleep(.2)
    sys.stdout.close()
    return i + y

seq = [(100, 4, 'task1.log'), (100, 5, 'task2.log')]
processes = 2
print ("Multiprocessing...")
pool = mp.Pool(processes)
resultsObj = pool.starmap_async(my_func, seq )          
pool.close()
results = resultsObj.get()

实时监控方式：

• Linux/macOS：打开两个终端，分别执行 tail -f task1.log 和 tail -f task2.log
• Windows：用PowerShell执行 Get-Content task1.log -Wait，或用支持实时刷新的文本编辑器打开日志文件

方法3：用日志模块实现规范的独立日志 #

使用Python内置的logging模块，给每个进程配置独立的日志处理器，相比直接重定向输出更稳定，还能添加时间戳等元信息。

修改后的代码示例：

import pathos.multiprocessing as mp
from time import sleep
import logging

def setup_logger(log_file):
    # 为每个任务创建独立日志实例
    logger = logging.getLogger(f"task_{log_file}")
    logger.setLevel(logging.INFO)
    handler = logging.FileHandler(log_file)
    formatter = logging.Formatter('%(asctime)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    return logger

def my_func(x, y, log_file):
    logger = setup_logger(log_file)
    for i in range(x):
        logger.info(y+i)
        sleep(.2)
    return i + y

seq = [(100, 4, 'task1.log'), (100, 5, 'task2.log')]
processes = 2
print ("Multiprocessing...")
pool = mp.Pool(processes)
resultsObj = pool.starmap_async(my_func, seq )          
pool.close()
results = resultsObj.get()

日志文件内容示例：

2024-05-20 14:30:00,123 - 4
2024-05-20 14:30:00,324 - 5

同样可以用终端命令实时监控不同日志文件的更新。

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