栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 面试经验 > 面试问答

子流程中有多个管道

面试问答 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

子流程中有多个管道

模拟bash进程替换:

#!/usr/bin/env pythonfrom subprocess import check_callcheck_call('someprogram <(someprocess) <(anotherprocess)',shell=True, executable='/bin/bash')

在Python中,您可以使用命名管道:

#!/usr/bin/env pythonfrom subprocess import Popenwith named_pipes(n=2) as paths:    someprogram = Popen(['someprogram'] + paths)    processes = []    for path, command in zip(paths, ['someprocess', 'anotherprocess']):        with open(path, 'wb', 0) as pipe: processes.append(Popen(command, stdout=pipe, close_fds=True))    for p in [someprogram] + processes:        p.wait()

在哪里

named_pipes(n)

import osimport shutilimport tempfilefrom contextlib import contextmanager@contextmanagerdef named_pipes(n=1):    dirname = tempfile.mkdtemp()    try:        paths = [os.path.join(dirname, 'named_pipe' + str(i)) for i in range(n)]        for path in paths: os.mkfifo(path)        yield paths    finally:        shutil.rmtree(dirname)

实现bash进程替换的另一种更可取的方式(无需在磁盘上创建命名条目)是使用@Dunes建议的

/dev/fd/N
文件名(如果可用)。在FreeBSD上,()为进程打开的所有文件描述符创建条目。要测试可用性,请运行:
fdescfs(5)
/dev/fd/#

$ test -r /dev/fd/3 3</dev/null && echo /dev/fd is available

如果失败;尝试进行符号链接

/dev/fd
proc(5)
就像在某些Linux上所做的那样:

$ ln -s /proc/self/fd /dev/fd

这是

/dev/fd
基于
someprogram <(someprocess) <(anotherprocess)
bash命令的实现:

#!/usr/bin/env python3from contextlib import ExitStackfrom subprocess import CalledProcessError, Popen, PIPEdef kill(process):    if process.poll() is None: # still running        process.kill()with ExitStack() as stack: # for proper cleanup    processes = []    for command in [['someprocess'], ['anotherprocess']]:  # start child processes        processes.append(stack.enter_context(Popen(command, stdout=PIPE)))        stack.callback(kill, processes[-1]) # kill on someprogram exit    fds = [p.stdout.fileno() for p in processes]    someprogram = stack.enter_context(        Popen(['someprogram'] + ['/dev/fd/%d' % fd for fd in fds], pass_fds=fds))    for p in processes: # close pipes in the parent        p.stdout.close()# exit stack: wait for processesif someprogram.returnpre != 0: # errors shouldn't go unnoticed   raise CalledProcessError(someprogram.returnpre, someprogram.args)

注意:在我的Ubuntu机器上,该

subprocess
代码仅在Python 3.4+中有效,尽管
pass_fds
自Python 3.2起可用。



转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/388520.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号