如何在Python中异步运行多条Linux命令？报错求助

问题重现 #

尝试用Python异步运行一系列Linux命令，命令能正常打印，但调用asyncio.run时触发FileNotFoundError，报错信息显示系统找不到包含完整命令的"文件"。

原代码 #

import asyncio
import subprocess

def get_args():

    parser = argparse.ArgumentParser(
        description='Run bulk generator in Gen server')

    parser.add_argument('--number_of_intercept', default='200',
                help='Number of intercept going into MSOM')

    return parser.parse_args()

async def bulk_generator (get_args):
    outputlist = []

    generator = "/bin/generator/./Generator.exe --direth -w 0000:84:00.1 -- -c 500 -d 62,598,62,1500,62 -r 500 -g eth+vlan:vlan_stream-id="+str(get_args)+"+vlan:vlan_stream-id="+str(get_args)+",span_number_vlan_ids=1+binseqn stream-id=100,eth_src=11:22:33:44:55:66"
    print(generator)
    p = subprocess.Popen([generator], stdout=subprocess.PIPE)
    output = str(p.communicate())
    outputlist.append(output)
    return outputlist


if __name__ == "__main__":

    args = get_args()
    for i in range(1,int(args.number_of_intercept)):
      asyncio.run(bulk_generator(i))

错误输出 #

Traceback (most recent call last):
  File "/Users/jasonleung/generator.py", line 32, in <module>
    asyncio.run(bulk_generator(i))
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/Users/jasonleung/generator.py", line 22, in bulk_generator
    p = subprocess.Popen([generator], stdout=subprocess.PIPE)
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/subprocess.py", line 969, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/subprocess.py", line 1845, in _execute_child
    raise child_exception_type(errno_num, err_msg, err_filename)
FileNotFoundError: [Errno 2] No such file or directory: '/bin/generator/./Generator.exe --direth -w 0000:84:00.1 -- -c 500 -d 62,598,62,1500,62 -r 500 -g eth+vlan:vlan_stream-id=1+vlan:vlan_stream-id=1,span_number_vlan_ids=1+binseqn stream-id=100,eth_src=11:22:33:44:55:66'

Process finished with exit code 1

错误原因分析 #

• subprocess参数格式错误：subprocess.Popen([generator])把整个命令字符串作为列表唯一元素，系统会将这个完整字符串当作要执行的文件名，而非拆分命令路径和参数。
• 同步调用阻塞异步循环：使用同步的subprocess.Popen和p.communicate()会阻塞asyncio事件循环，完全失去异步的意义。
• 循环调用asyncio.run：每次循环创建新的事件循环，效率极低，不符合异步编程的最佳实践。
• 参数命名冲突：bulk_generator的参数名get_args和上方的函数名重名，易引发混淆。

修复方案 #

1. 拆分命令为路径+参数列表：将命令拆分为可执行文件路径和独立的参数列表，避免系统误判文件名。
2. 使用异步subprocess接口：改用asyncio.create_subprocess_exec执行异步子进程，配合communicate()方法异步获取输出。
3. 批量管理异步任务：收集所有异步任务，一次性通过asyncio.gather执行，避免重复创建事件循环。
4. 修正参数命名：将bulk_generator的参数名改为stream_id，避免与函数名冲突。

修复后完整代码 #

import asyncio
import subprocess
from argparse import ArgumentParser

def get_args():
    parser = ArgumentParser(description='Run bulk generator in Gen server')
    parser.add_argument('--number_of_intercept', default='200',
                        help='Number of intercept going into MSOM')
    return parser.parse_args()

async def bulk_generator(stream_id):
    output_list = []
    # 拆分命令为路径和参数列表
    cmd_path = "/bin/generator/Generator.exe"
    cmd_args = [
        "--direth", "-w", "0000:84:00.1", "--",
        "-c", "500", "-d", "62,598,62,1500,62", "-r", "500",
        "-g", f"eth+vlan:vlan_stream-id={stream_id}+vlan:vlan_stream-id={stream_id},span_number_vlan_ids=1+binseqn stream-id=100,eth_src=11:22:33:44:55:66"
    ]
    # 打印完整命令
    print(f"{cmd_path} {' '.join(cmd_args)}")
    
    # 使用异步subprocess执行命令
    process = await asyncio.create_subprocess_exec(
        cmd_path, *cmd_args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    
    # 处理输出
    output = stdout.decode('utf-8') + stderr.decode('utf-8')
    output_list.append(output)
    return output_list

if __name__ == "__main__":
    args = get_args()
    # 收集所有异步任务
    tasks = [bulk_generator(i) for i in range(1, int(args.number_of_intercept))]
    # 一次性运行所有任务
    asyncio.run(asyncio.gather(*tasks))

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