利用這兩個特點，作者巧妙地實現(xiàn)了程序邏輯。

第二個值得注意的地方，是在 run() 函數(shù)的內(nèi)部，嵌套定義了 target() 函數(shù)。

def target():

    try:
        self.process = subprocess.Popen(self.cmd,
            universal_newlines=True,
            shell=False,
            env=environ,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=0,
            cwd=cwd,
        )

        if sys.version_info[0] >= 3:
            self.out, self.err = self.process.communicate(
                input = bytes(self.data, "UTF-8") if self.data else None
            )
        else:
            self.out, self.err = self.process.communicate(self.data)
    except Exception as exc:
        self.exc = exc

在 Python 中，函數(shù)定義是允許嵌套的，不過

• 各個函數(shù)有自己的作用域；
• 內(nèi)層函數(shù)優(yōu)先訪問內(nèi)層作用域的變量，如果內(nèi)層沒有所需變量，則逐層向外尋找所需變量；
• 外層函數(shù)不能訪問內(nèi)層函數(shù)的變量（對外層函數(shù)來說，這是局部變量）；除非內(nèi)層函數(shù)聲明變量時加上了 global 關(guān)鍵字修飾，并且在訪問它時已經(jīng)調(diào)用過內(nèi)層函數(shù)。

這里的 target() 函數(shù)定義了我們接到一個執(zhí)行 Shell 命令的需求時，我們要做哪些事情。依其定義，我們要首先使用 subprocess.Popen() 創(chuàng)建一個子進程，并在相應的條件下執(zhí)行 self.cmd。然后調(diào)用 self.process.communicate() 方法，將 self.data 通過管道傳給正在 Shell 中執(zhí)行的程序，并獲取程序的標準輸出和標準錯誤。在整個過程中，但凡出現(xiàn)任何問題，都保存在 self.exc 當中。這里作者使用了所有異常的基類 Exception，這是因為對于作者來說 self.cmd 是不可控的，在執(zhí)行 self.cmd 的過程中可能出現(xiàn)任何形式的異常。為了能夠處理所有異常，作者必須使用 Exception 來處理。

第三個值得注意的地方，是作者在工作線程中去實際執(zhí)行 target() 完成的任務。

if self.exc:
    raise self.exc
if _is_alive(thread) :
    _terminate_process(self.process)
    thread.join(kill_timeout)
    if _is_alive(thread):
        _kill_process(self.process)
        thread.join()
self.returncode = self.process.returncode
return self.out, self.err

首先，作者創(chuàng)建了一個線程，將 target() 函數(shù)作為參數(shù)傳入構(gòu)造。也就是說，thread.start() 實際會執(zhí)行 target() 函數(shù)的代碼。而后，作者用 thread.join(timeout) 的方式，來處理上層傳下來的超時限制。這樣，主進程將會阻塞住，直到

• 線程中的任務完成（也就是 target() 中創(chuàng)建的子進程的任務完成）；或者
• 達到超時時間限制。

第四個值得注意的地方，是作者回收和處理在線程中運行的子進程任務的執(zhí)行狀態(tài)信息。

if self.exc:
    raise self.exc
if _is_alive(thread) :
    _terminate_process(self.process)
    thread.join(kill_timeout)
    if _is_alive(thread):
        _kill_process(self.process)
        thread.join()
self.returncode = self.process.returncode
return self.out, self.err

首先，子進程可能拋出異常，因此需要捕獲和繼續(xù)向上拋出異常。

其次，線程 thread 可能因為超時而執(zhí)行到當前代碼，因此通過預定義的 _is_alive() 函數(shù)來判斷線程是正常退出還是扔在超時運行。如果確實超時，那么首先應該終止子進程，然后嘗試等待線程超時終止。如果線程仍然還活著，說明線程內(nèi)的子進程沒有被正確終止，那么首先殺死子進程，然后阻塞線程直到它完成。這樣的設計，是確保子進程和線程都完全停止，防止僵尸進程的出現(xiàn)。

最后，函數(shù)返回標準輸出和標準錯誤的內(nèi)容。

Response 類

我們來看 Response 類。這是一個模塊內(nèi)部使用的類，Response 類的每個實例都是 Command 類的實例調(diào)用 run() 方法后的執(zhí)行結(jié)果信息。

class Response(object):
    """A command's response"""

    def __init__(self, process=None):
        super(Response, self).__init__()

        self._process = process
        self.command = None
        self.std_err = None
        self.std_out = None
        self.status_code = None
        self.history = []


    def __repr__(self):
        if len(self.command):
            return '<Response [{0}]>'.format(self.command[0])
        else:
            return '<Response>'

從只有一個 __repr__() 方法可以看出，Response 類幾乎只是一個簡單的數(shù)據(jù)結(jié)構(gòu)，提供了可供打印的功能，僅此而已。那么作者為什么要設計這樣一個類呢？這里我們留給讀者思考。

expand_args 函數(shù)

expand_args(command) 函數(shù)接收一個字符串作為參數(shù)，并將之解析為一個個的命令。

Prepare arguments.

if isinstance(command, (str, unicode)):
splitter = shlex.shlex(command.encode('utf-8'))
splitter.whitespace = '|'
splitter.whitespace_split = True
command = []

while True:
token = splitter.get_token()
if token:
command.append(token)
else:
break

command = list(map(shlex.split, command))

return command

我們以 'cat inFile | sort | uniq' 為引數(shù)，傳入 expand_args 函數(shù)，分析一下會發(fā)生什么。

首先，作者用 shlex.shlex() 構(gòu)造了一個詞法分析器，并設置以管道符號 | 為標志，分割傳入的字符串（或者 unicode 類型的實例，后不再重復）。加上之后的 while 循環(huán)，這基本相當于執(zhí)行了 command = command.split('|') 的效果。

而后，執(zhí)行 command = list(map(shlex.split, command))，調(diào)用 shlex.split 函數(shù)，作用在 command 的每一個元素上，并返回一個列表，保存在 command 當中。最后以 return 將 command 返回給調(diào)用函數(shù)。

這里的 map() 函數(shù)接收兩個參數(shù)

• 一個函數(shù)
• 一個可迭代的列表

然后將函數(shù)作用在列表的每一個元素上，并返回一個列表。類似的函數(shù)還有 reduce() 函數(shù)（參考 Google 的 MapReduce 架構(gòu)）。這里給出兩個示例，供體會它們的作用

map.py

#!/usr/bin/env python
inIter = ['adam', 'LISA', 'barT']
regNames = lambda iter: map ((lambda inStr: inStr.capitalize()), iter)
print regNames (inIter)

最后，輸入 'cat inFile | sort | uniq' 有輸出 [['cat', 'inFile'], ['sort'], ['uniq']]。

run 函數(shù)

run(command, data=None, timeout=None, kill_timeout=None, env=None, cwd=None) 函數(shù)是 envoy 模塊的主要接口，用來在子進程里執(zhí)行 Shell 命令。

首先解釋一下 run() 函數(shù)的各個參數(shù)的含義

• command 需要執(zhí)行的 Shell 命令（可以包含管道，但是不允許包含 && 或者 ; 之類的符號）；
• data 通過管道傳入 Shell 命令的內(nèi)容；
• timeout 子進程執(zhí)行超時時間；
• kill_timeout 終止子進程失敗的超時時間，超過這個時間將直接殺死子進程；
• env 環(huán)境變量；
• cwd Current Working Directory，工作目錄。

run 函數(shù)的實現(xiàn)相對來說是平鋪直敘的，這里用注釋簡單說明一下各個部分都做了什么即可。

'''對 command 做詞法分析，拆分命令'''
command = expand_args(command)

history = []
for c in command:

'''模擬管道的作用，傳入上一程序的標準輸出'''
if len(history):

due to broken pipe problems pass only first 10 KiB

data = history[-1].std_out[0:10*1024]

'''實際在子進程里執(zhí)行命令'''
cmd = Command(c)
try:
out, err = cmd.run(data, timeout, kill_timeout, env, cwd)
status_code = cmd.returncode
except OSError as e:
out, err = '', u"\n".join([e.strerror, traceback.format_exc()])
status_code = 127

'''將執(zhí)行結(jié)果保存在 history 當中'''
r = Response(process=cmd)

r.command = c
r.std_out = out
r.std_err = err
r.status_code = status_code

history.append(r)

'''函數(shù)返回最后一個管道（如果有）之后命令的輸出和詳細情況'''
r = history.pop()
r.history = history

return r
</pre>

|

模塊設計分析

Kenneth Reitz 不愧是公認的這個世界上 Python 代碼寫得最好的人之一——雖然 envoy 是對 subprocess 的簡單封裝，功能有限，但是代碼結(jié)構(gòu)非常優(yōu)雅，內(nèi)部實現(xiàn)的逐層封裝十分完善。

對于模塊的用戶（程序員）來說，envoy 幾乎只有 run 這一個入口。而它的作用也很明確：開一個子進程，執(zhí)行一條 Shell 命令，然后在規(guī)定時間內(nèi)取得執(zhí)行結(jié)果——中間的臟活累活（處理異常、超時、主進程阻塞等待、保存歷史等等），envoy 都幫你做好了。

對于 run() 函數(shù)來說，它只需要知道執(zhí)行 out, err = cmd.run() 就能在子進程里執(zhí)行用戶需要的命令，然后將結(jié)果存在 Response 里就可以了。

對于 Command.run() 函數(shù)來說，它只需要處理好環(huán)境變量，執(zhí)行 target() 最后處理超時、異常、收集結(jié)果信息就可以了。

對于 target() 來說，這是一個嵌套定義的函數(shù)，它才是真正 fork 子進程并執(zhí)行 Shell 命令的函數(shù)。

不難發(fā)現(xiàn)，每個層次完成的任務，幾乎都可以用簡單一句話解釋清楚

• envoy.run() - 將 Shell 命令交給它，就會在子進程里執(zhí)行這些命令并處理好返回結(jié)果；
• Command.run() - 將一個具體的 Shell 命令（不包含管道）交給它，就會在子進程里執(zhí)行這些命令并處理好返回結(jié)果；
• target() - fork 一個子進程，然后在子進程里開心地執(zhí)行命令。

這種符合 *nix 哲學的設計，造就了優(yōu)雅好用的 envoy 庫。對于程序員來說，將命令交給它，然后坐等結(jié)果就可以了。無愧于 Python Subprocesses for Humans 的豪言壯語。

轉(zhuǎn)自：
https://blog.csdn.net/junli_chen/article/details/78295454