如何通过PIPE并行运行Python Subprocess？（SVG转PNG场景）

我正在使用Inkscape将大量SVG图片转换为PNG格式。

单线程实现 #

单线程代码如下，转换30张图片耗时85秒：

import subprocess
import time
import os

inkscape_path = r'C:\Program Files\Inkscape\bin\inkscape.com'

steps=30
filenames = []
processes = []

t_start = time.time()
for i in range(steps):
    template = bytes(f"""<?xml version="1.0" encoding="UTF-8"?>
    <svg width="100" height="100" viewBox="-10 -10 30 30" xmlns="http://www.w3.org/2000/svg">                   
    <line x1="0" y1="0" x2="{i*9/(steps-1)+1}" y2="0" stroke="green"/>
    </svg>
    """,'UTF-8')

    filename = f'img{i:02}.png'
    filenames.append(filename)

    process = subprocess.run([inkscape_path, '-p', '-o', filename],input=template )

elapsed_time = time.time() - t_start
print(elapsed_time)

不规范的并行尝试 #

我尝试用communicate()配合超时忽略异常的方式实现并行，耗时30秒，但这种写法不够规范：

import subprocess
import time
import os

inkscape_path = r'C:\Program Files\Inkscape\bin\inkscape.com'

steps=30
filenames = []
processes = []

t_start = time.time()
for i in range(steps):
    template = bytes(f"""<?xml version="1.0" encoding="UTF-8"?>
    <svg width="100" height="100" viewBox="-10 -10 30 30" xmlns="http://www.w3.org/2000/svg">                   
    <line x1="0" y1="0" x2="{i*9/(steps-1)+1}" y2="0" stroke="green"/>
    </svg>
    """,'UTF-8')

    filename = f'img{i:02}.png'
    filenames.append(filename)

    process = subprocess.Popen([inkscape_path, '-p', '-o', filename],stdin= subprocess.PIPE, env=dict(os.environ, SELF_CALL="xxx") )

    try:
        process.communicate(template,timeout=0)
    except:
        pass
   
    processes.append(process)
    
for p in processes:
    p.wait()

elapsed_time = time.time() - t_start
print(elapsed_time)

请问如何以更简洁规范的方式并行运行这些子进程？最好能通过一行代码实现。

（注：SELF_CALL是临时解决方案，用于规避Inkscape多实例运行时偶尔抛出的异常）

针对调用外部进程这类IO密集型任务，推荐用concurrent.futures.ThreadPoolExecutor实现高效且规范的并行，线程池开销远低于进程池，代码结构清晰。

简洁实现（核心并行逻辑一行） #

把转换逻辑封装为函数，通过线程池批量提交任务：

import subprocess
import time
import os
from concurrent.futures import ThreadPoolExecutor

inkscape_path = r'C:\Program Files\Inkscape\bin\inkscape.com'
steps = 30

def convert_svg(i):
    template = bytes(f"""<?xml version="1.0" encoding="UTF-8"?>
    <svg width="100" height="100" viewBox="-10 -10 30 30" xmlns="http://www.w3.org/2000/svg">                   
    <line x1="0" y1="0" x2="{i*9/(steps-1)+1}" y2="0" stroke="green"/>
    </svg>
    """, 'UTF-8')
    filename = f'img{i:02}.png'
    env = dict(os.environ, SELF_CALL="xxx")
    subprocess.run([inkscape_path, '-p', '-o', filename], input=template, env=env)

t_start = time.time()
# 核心并行逻辑：自动调度任务，max_workers按需调整
with ThreadPoolExecutor(max_workers=os.cpu_count() * 2) as executor:
    executor.map(convert_svg, range(steps))  # 这一行完成并行调度

elapsed_time = time.time() - t_start
print(elapsed_time)

优势说明 #

• 无需手动管理进程生命周期，线程池自动处理任务调度和资源回收
• 可轻松添加异常捕获逻辑，避免原写法中忽略所有异常的问题
• max_workers可根据系统资源灵活调整，防止资源过载
• 代码结构清晰，可读性强，维护成本低

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