詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

來(lái)源：懂視網(wǎng) 責(zé)編：小采時(shí)間：2020-11-27 14:16:16

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法:這里我們用Windows下的shell來(lái)舉例： subprocess *為了方便你理解，我們用一個(gè)很簡(jiǎn)單的一段代碼來(lái)說(shuō)明：可以看見(jiàn)我們利用Popen實(shí)例化了一個(gè)p，創(chuàng)建了子程序cmd.exe，然后我們給他的的Stdin（標(biāo)準(zhǔn)輸入流）Stdout（標(biāo)準(zhǔn)輸出流）;同時(shí)使用了subproce

推薦度：

點(diǎn)擊下載本文 文檔為doc格式

導(dǎo)讀詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法:這里我們用Windows下的shell來(lái)舉例： subprocess *為了方便你理解，我們用一個(gè)很簡(jiǎn)單的一段代碼來(lái)說(shuō)明：可以看見(jiàn)我們利用Popen實(shí)例化了一個(gè)p，創(chuàng)建了子程序cmd.exe，然后我們給他的的Stdin（標(biāo)準(zhǔn)輸入流）Stdout（標(biāo)準(zhǔn)輸出流）;同時(shí)使用了subproce

這里我們用Windows下的shell來(lái)舉例：

 subprocess *

為了方便你理解，我們用一個(gè)很簡(jiǎn)單的一段代碼來(lái)說(shuō)明：

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

可以看見(jiàn)我們利用Popen實(shí)例化了一個(gè)p，創(chuàng)建了子程序cmd.exe，然后我們給他的的Stdin（標(biāo)準(zhǔn)輸入流）Stdout（標(biāo)準(zhǔn)輸出流）;

同時(shí)使用了subprocess.PIPE 作為參數(shù)，這個(gè)是一個(gè)特殊值，用于表明這些通道要開(kāi)放。(在Python3.5，加入了run()方法來(lái)進(jìn)行更好的操作)

然后我們繼續(xù)

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

這些信息是不是很眼熟？這都是cmd的標(biāo)準(zhǔn)輸出！

然后就會(huì)輸出這些：

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

我們剛剛所寫(xiě)入的信息"echo Hellwworlds "已經(jīng)被寫(xiě)入了，看起來(lái)確實(shí)成功了！

注意一下我們使用了 p.stdin.flush() 來(lái)對(duì)輸入緩存區(qū)進(jìn)行刷新，輸出的信息也需要一個(gè) " "，至少在 Windows 系統(tǒng)下必須這樣做，否則只刷新（p.stdin.flush）的話是無(wú)效的；

我們到底做了什么？

我們成功的創(chuàng)建了子程序 cmd.exe，并且寫(xiě)入"echo Hellwworlds " ，然后cmd獲取了并且執(zhí)行，于是返回 Hellwworlds，這就是一次很簡(jiǎn)單的讀寫(xiě)交互！

更高級(jí)的使用

既然我們已經(jīng)可以簡(jiǎn)單的讀寫(xiě)了，那么加點(diǎn)for和threading 吧，味道也許更佳喔~

#run.py 

from subprocess import * 
import threading 
import time

p =Popen('cmd.exe',shell=True,stdin=PIPE,stdout=PIPE)

def run():
 global p
 while True:
 line = p.stdout.readline() 
 if not line: #空則跳出
 break
 print(">>>>>>",line.decode("GBK"))

 print("look up!!! EXIT ===") #跳出


w =threading.Thread(target=run)

p.stdin.write("echo HELLW_WORLD!
".encode("GBK"))
p.stdin.flush()
time.sleep(1) #延遲是因?yàn)榈却幌戮€程就緒
p.stdin.write("exit
".encode("GBK"))
p.stdin.flush()

w.start()

很好很好，猜猜輸出什么？

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法

有很多換行的原因是cmd返回的結(jié)果有換行，然后print輸出會(huì)加一個(gè)換行，所以就換了兩行，你可以考慮使用 sys.stdout.write 來(lái)輸出，這樣就沒(méi)有附加的換行了

這樣的話，你可以制作一個(gè)基礎(chǔ)的讀寫(xiě)了，那么我們開(kāi)始封裝吧。

封裝Pipe

不廢話了，直接上代碼，如果你真的想學(xué)會(huì)的話，還請(qǐng)認(rèn)真自己讀讀代碼。

110行

我們實(shí)現(xiàn)了將所有的過(guò)程集中在一個(gè)類里面，并且可以定義三個(gè)參數(shù)，退出反饋函數(shù)，就緒反饋函數(shù)和輸出反饋函數(shù)。

# -*- coding:utf-8 -*-

import subprocess 
import sys
import threading

class LoopException(Exception):
 """循環(huán)異常自定義異常，此異常并不代表循環(huán)每一次都是非正常退出的"""
 def __init__(self,msg="LoopException"):
 self._msg=msg

 def __str__(self):
 return self._msg



class SwPipe():
 """
 與任意子進(jìn)程通信管道類，可以進(jìn)行管道交互通信
 """
 def __init__(self,commande,func,exitfunc,readyfunc=None,
 shell=True,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,code="GBK"):
 """
 commande 命令
 func 正確

輸出反饋函數(shù) exitfunc 異常反饋函數(shù) readyfunc 當(dāng)管道創(chuàng)建完畢時(shí)調(diào)用 """ self._thread = threading.Thread(target=self.__run,args=(commande,shell,stdin,stdout,stderr,readyfunc)) self._code = code self._func = func self._exitfunc = exitfunc self._flag = False self._CRFL = " " def __run(self,commande,shell,stdin,stdout,stderr,readyfunc): """ 私有函數(shù) """ try: self._process = subprocess.Popen( commande, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr ) except OSError as e: self._exitfunc(e) fun = self._process.stdout.readline self._flag = True if readyfunc != None: threading.Thread(target=readyfunc).start() #準(zhǔn)備就緒 while True: line = fun() if not line: break try: tmp = line.decode(self._code) except UnicodeDecodeError: tmp = self._CRFL + "[PIPE_CODE_ERROR] <Code ERROR: UnicodeDecodeError> " + "[PIPE_CODE_ERROR] Now code is: " + self._code + self._CRFL self._func(self,tmp) self._flag = False self._exitfunc(LoopException("While Loop break")) #正常退出 def write(self,msg): if self._flag: #請(qǐng)注意一下這里的換行 self._process.stdin.write((msg + self._CRFL).encode(self._code)) self._process.stdin.flush() #sys.stdin.write(msg)#怎么說(shuō)呢，無(wú)法直接用代碼發(fā)送指令，只能默認(rèn)的stdin else: raise LoopException("Shell pipe error from '_flag' not True!") #還未準(zhǔn)備好就退出 def start(self): """ 開(kāi)始線程 """ self._thread.start() def destroy(self): """ 停止并銷毀自身 """ process.stdout.close() self._thread.stop() del self if __name__ == '__main__': #那么我們來(lái)開(kāi)始使用它吧 e = None #反饋函數(shù) def event(cls,line):#輸出反饋函數(shù) sys.stdout.write(line) def exit(msg):#退出反饋函數(shù) print(msg) def ready():#線程就緒反饋函數(shù) e.write("dir") #執(zhí)行 e.write("ping www.baidu.com") e.write("echo Hello!World 你好中國(guó)！你好世界！") e.write("exit") e = SwPipe("cmd.exe",event,exit,ready) e.start()

輸出：

詳解Python3利subprocess實(shí)現(xiàn)管道pipe交互操作讀/寫(xiě)通信方法