錯(cuò)誤、調(diào)試和測試

程序運(yùn)行中，可能會(huì)遇到BUG、用戶輸入異常數(shù)據(jù)以及其它環(huán)境的異常，這些都需要程序猿進(jìn)行處理。Python提供了一套內(nèi)置的異常處理機(jī)制，供程序猿使用，同時(shí)PDB提供了調(diào)試代碼的功能，除此之外，程序猿還應(yīng)該掌握測試的編寫，確保程序的運(yùn)行符合預(yù)期。

錯(cuò)誤處理

在一般程序處理中，可以對(duì)函數(shù)的返回值進(jìn)行檢查，是否返回了約定的錯(cuò)誤碼。例如系統(tǒng)程序調(diào)用的錯(cuò)誤碼一般都是-1，成功返回0。但是這種方式必須用大量的代碼來判斷是否出錯(cuò)，所以高級(jí)語言內(nèi)置了try...except...finally的錯(cuò)誤機(jī)制。

try:print('try...')r = 10 / int('2')print('result:', r) except ValueError as e:print('ValueError:', e) except ZeroDivisionError as e:print('ZeroDivisionError:', e) else:print('no error!') finally:print('finally...') print(‘END’）

當(dāng)我們認(rèn)為某些代碼可能會(huì)出錯(cuò)時(shí)，就可以用try來運(yùn)行這段代碼，如果執(zhí)行出錯(cuò)，則后續(xù)代碼不會(huì)繼續(xù)執(zhí)行，而是直接跳轉(zhuǎn)至錯(cuò)誤處理代碼，即except語句塊，執(zhí)行完except后，如果有finally語句塊，則執(zhí)行finally語句塊，至此，執(zhí)行完畢。如果發(fā)生了不同的錯(cuò)誤類型，可以由不同的except語句塊處理，可以沒有finally語句塊。

Python的錯(cuò)誤也是類，所有的錯(cuò)誤類型都繼承自BaseException，常見的錯(cuò)誤類型和繼承關(guān)系參考?官方文檔

使用try...except捕獲錯(cuò)誤還有一個(gè)巨大的好處，就是可以跨越多層調(diào)用，比如函數(shù)main()調(diào)用foo()，foo()調(diào)用bar()，結(jié)果bar()出錯(cuò)了，這時(shí)，只要main()捕獲到了，就可以處理。

調(diào)用堆棧

如果錯(cuò)誤沒有被捕獲，它就會(huì)一直往上拋，最后被Python解釋器捕獲，打印一個(gè)錯(cuò)誤信息，然后程序退出。出錯(cuò)并不可怕，可怕的是不知道哪里出錯(cuò)了。解讀錯(cuò)誤信息是定位錯(cuò)誤的關(guān)鍵。

記錄錯(cuò)誤

在C語言中，如果發(fā)生錯(cuò)誤想要記錄，必須自己編寫錯(cuò)誤記錄的程序。Python內(nèi)置的logging模塊可以非常容易地記錄錯(cuò)誤信息。通過配置，logging還可以把錯(cuò)誤記錄到日志文件里，方便事后排查。

# err_logging.pyimport loggingdef foo(s):return 10 / int(s)def bar(s):return foo(s) * 2def main():try:bar('0')except Exception as e:logging.exception(e)main() print('END')

拋出錯(cuò)誤

拋出錯(cuò)誤，首先需要定義一個(gè)錯(cuò)誤 Class，選擇好繼承關(guān)系，然后用raise語句拋出一個(gè)錯(cuò)誤實(shí)例。如果可以盡量使用Python內(nèi)置的錯(cuò)誤類型，僅在非常必要的時(shí)候自己定義錯(cuò)誤類。

# err_raise.py class FooError(ValueError):passdef foo(s):n = int(s)if n==0:raise FooError('invalid value: %s' % s)return 10 / nfoo('0')

調(diào)試 Debug

調(diào)試最簡單的辦法就是print()，這個(gè)方法最簡單，但是在發(fā)布的時(shí)候需要把所有的調(diào)試信息注釋掉。

斷言 assert

凡是用print()來輔助查看的地方，都可以用斷言（assert）來替代。

def foo(s):n = int(s)assert n != 0, 'n is zero!'return 10 / ndef main():foo('0')

assert的意思是，表達(dá)式n != 0應(yīng)該是True，否則，根據(jù)程序運(yùn)行的邏輯，后面的代碼肯定會(huì)出錯(cuò)。如果斷言失敗，assert語句本身就會(huì)拋出AssertionError。

啟動(dòng)Python解釋器時(shí)可以用-O參數(shù)來關(guān)閉assert。

logging

使用 logging 不僅可以拋出錯(cuò)誤，還可以輸出到文件。

import logging logging.basicConfig(level=logging.INFO)s = '0' n = int(s) logging.info('n = %d' % n) print(10 / n)

這就是logging的好處，它允許你指定記錄信息的級(jí)別，有debug，info，warning，error等幾個(gè)級(jí)別，當(dāng)我們指定level=INFO時(shí)，logging.debug就不起作用了。同理，指定level=WARNING后，debug和info就不起作用了。這樣一來，你可以放心地輸出不同級(jí)別的信息，也不用刪除，最后統(tǒng)一控制輸出哪個(gè)級(jí)別的信息。

logging的另一個(gè)好處是通過簡單的配置，一條語句可以同時(shí)輸出到不同的地方，比如console和文件。

pdb

可以在命令行下使用pdb,啟動(dòng)Python的調(diào)試器pdb，讓程序以單步方式運(yùn)行，可以隨時(shí)查看運(yùn)行狀態(tài)。

# err.py s = '0' n = int(s) print(10 / n)$ python3 -m pdb err.py > /Users/michael/Github/learn-python3/samples/debug/err.py(2)<module>() -> s = ‘0'

輸入1可以查看代碼，輸入n可以單步執(zhí)行代碼。使用p來查看變量，使用q退出調(diào)試。

pdb.set_trace()

這個(gè)方法也是用pdb，但是不需要單步執(zhí)行，我們只需要import pdb，然后，在可能出錯(cuò)的地方放一個(gè)pdb.set_trace()，就可以設(shè)置一個(gè)斷點(diǎn)。運(yùn)行代碼，程序會(huì)自動(dòng)在pdb.set_trace()暫停并進(jìn)入pdb調(diào)試環(huán)境，可以用命令p查看變量，或者用命令c繼續(xù)運(yùn)行。

單元測試

單元測試是用來對(duì)一個(gè)模塊、一個(gè)函數(shù)或者一個(gè)類來進(jìn)行正確性檢驗(yàn)的測試工作。

文檔測試

doctest非常有用，不但可以用來測試，還可以直接作為示例代碼。通過某些文檔生成工具，就可以自動(dòng)把包含doctest的注釋提取出來。用戶看文檔的時(shí)候，同時(shí)也看到了doctest。

IO編程

IO就是Input / Output ，也就是輸入和輸出。IO編程中，Stream（流）是一個(gè)很重要的概念，可以把流想象成一個(gè)水管，數(shù)據(jù)就是水管里的水，但是只能單向流動(dòng)。

由于計(jì)算機(jī)各個(gè)部件之間的速度不一致，所以處理IO問題時(shí)有兩種辦法：同步IO、異步IO。同步和異步的區(qū)別就在于是否等待IO執(zhí)行的結(jié)果。

文件讀寫

讀寫文件是最常見的IO操作。Python內(nèi)置了讀寫文件的函數(shù)，用法和C是兼容的。在磁盤上讀寫文件的功能都是由操作系統(tǒng)提供的，現(xiàn)代操作系統(tǒng)不允許普通的程序直接操作磁盤，所以，讀寫文件就是請求操作系統(tǒng)打開一個(gè)文件對(duì)象（通常稱為文件描述符），然后，通過操作系統(tǒng)提供的接口從這個(gè)文件對(duì)象中讀取數(shù)據(jù)（讀文件），或者把數(shù)據(jù)寫入這個(gè)文件對(duì)象（寫文件）。

讀文件

try:f = open('/path/to/file', 'r')print(f.read()) finally:if f:f.close()with open('/path/to/file', 'r') as f:print(f.read())

類似于c語言，open函數(shù)默認(rèn)接收一個(gè)文件名、一個(gè)打開模式參數(shù)（r、w默認(rèn)對(duì)應(yīng)文本文件，rb對(duì)應(yīng)二進(jìn)制文件）。默認(rèn)打開的是UTF-8編碼的文件，如果需要打開其它編碼的，需要傳入encoding參數(shù)，如果文本的編碼不一致可能導(dǎo)致讀取出錯(cuò)，可以傳入錯(cuò)誤處理參數(shù)errors。read方法一次將文件的所有內(nèi)容讀入內(nèi)存，可以通過參數(shù)指定讀入的長度read(size)，也可以使用readline方法每次讀入一行，使用readlines一次讀入所有的行。文件使用后注意要進(jìn)行關(guān)閉。

寫文件

>>> f = open('/Users/michael/test.txt', 'w') >>> f.write('Hello, world!') >>> f.close()with open('/Users/michael/test.txt', 'w') as f:f.write('Hello, world!’)

寫文件和讀文件是一樣的，唯一區(qū)別是調(diào)用open()函數(shù)時(shí)，傳入標(biāo)識(shí)符’w’或者’wb’表示寫文本文件或?qū)懚M(jìn)制文件。當(dāng)我們寫文件時(shí)，操作系統(tǒng)往往不會(huì)立刻把數(shù)據(jù)寫入磁盤，而是放到內(nèi)存緩存起來，空閑的時(shí)候再慢慢寫入。只有調(diào)用close()方法時(shí)，操作系統(tǒng)才保證把沒有寫入的數(shù)據(jù)全部寫入磁盤。

StringIO 和 BytesIO

很多時(shí)候，數(shù)據(jù)讀寫不一定是文件，也可以在內(nèi)存中讀寫。StringIO顧名思義就是在內(nèi)存中讀寫str。

>>> from io import StringIO >>> f = StringIO() >>> f.write('hello') 5 >>> f.write(' ') 1 >>> f.write('world!') 6 >>> print(f.getvalue()) hello world!>>> from io import StringIO >>> f = StringIO('Hello!\nHi!\nGoodbye!') >>> while True: ... s = f.readline() ... if s == '': ... break ... print(s.strip()) ... Hello! Hi! Goodbye!

StringIO操作的只能是str，如果要操作二進(jìn)制數(shù)據(jù)，就需要使用BytesIO。BytesIO實(shí)現(xiàn)了在內(nèi)存中讀寫bytes。

操作文件和目錄

Python內(nèi)置的os模塊也可以直接調(diào)用操作系統(tǒng)提供的接口函數(shù)。import os模塊后，就可以調(diào)用一些系統(tǒng)命令。

>>> import os >>> os.name # 操作系統(tǒng)類型 'posix' >>> os.uname() posix.uname_result(sysname='Darwin', nodename='RousseaudeMacBook-Pro.local', release='15.6.0', version='Darwin Kernel Version 15.6.0: Mon Jan 9 23:07:29 PST 2017; root:xnu-3248.60.11.2.1~1/RELEASE_X86_64', machine='x86_64') >>> os.environ environ({'TERM_PROGRAM': 'Apple_Terminal', 'SHELL': '/bin/bash', 'TERM': 'xterm-256color', 'TMPDIR': '/var/folders/95/zrdts1md6j942mpyd7kd875h0000gn/T/', 'Apple_PubSub_Socket_Render': '/private/tmp/com.apple.launchd.fhDfjTsyk6/Render', 'TERM_PROGRAM_VERSION': '361.1', 'OLDPWD': '/Users/rousseau/Projects/python.my', 'TERM_SESSION_ID': '5A1B275C-3BE5-4673-B163-29DFF5C19C77', 'USER': 'rousseau', 'SSH_AUTH_SOCK': '/private/tmp/com.apple.launchd.mLtAPJeOFm/Listeners', '__CF_USER_TEXT_ENCODING': '0x1F5:0x0:0x0', 'PATH': '/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin', 'PWD': '/Users/rousseau/Projects/python.my/mypython', 'XPC_FLAGS': '0x0', 'XPC_SERVICE_NAME': '0', 'SHLVL': '1', 'HOME': '/Users/rousseau', 'LOGNAME': 'rousseau', 'LC_CTYPE': 'UTF-8', '_': '/usr/local/bin/python3', '__PYVENV_LAUNCHER__': '/usr/local/bin/python3’}) >>> os.path.abspath('.') # 查看當(dāng)前目錄的絕對(duì)路徑: '/Users/rousseau/Projects/python.my/mypython’ # 在某個(gè)目錄下創(chuàng)建一個(gè)新目錄，首先把新目錄的完整路徑表示出來: >>> os.path.join('/Users/michael', 'testdir') '/Users/michael/testdir' # 然后創(chuàng)建一個(gè)目錄: >>> os.mkdir('/Users/michael/testdir') # 刪掉一個(gè)目錄: >>> os.rmdir('/Users/michael/testdir’)

因?yàn)閃indows和Unix的路徑表達(dá)方式不一樣，所以在處理路徑時(shí)，盡量使用Python提供的os.path.join()和os.path.split()避免處理發(fā)生問題。其它的文件處理函數(shù)os.rename、os.remove。

序列化

序列號(hào)我理解的就是將內(nèi)存中變量的狀態(tài)和值轉(zhuǎn)換為文本，以方便進(jìn)行持久化的存儲(chǔ)，也可能不進(jìn)行存儲(chǔ)，但是序列話之后方便進(jìn)行傳輸。我們把變量從內(nèi)存中變成可存儲(chǔ)或傳輸?shù)倪^程稱之為序列化，在Python中叫pickling，在其他語言中也被稱之為serialization，marshalling，flattening等等，都是一個(gè)意思。反過來，把變量內(nèi)容從序列化的對(duì)象重新讀到內(nèi)存里稱之為反序列化，即unpickling。

Python提供了pickle模塊來實(shí)現(xiàn)序列化。

>>> import pickle >>> d = dict(name='Bob', age=20, score=88) >>> pickle.dumps(d) b'\x80\x03}q\x00(X\x03\x00\x00\x00ageq\x01K\x14X\x05\x00\x00\x00scoreq\x02KXX\x04\x00\x00\x00nameq\x03X\x03\x00\x00\x00Bobq\x04u.’#也可以將序列化的內(nèi)容寫入文本 >>> f = open('dump.txt', 'wb') >>> pickle.dump(d, f) >>> f.close()#讀取的過程 >>> f = open('dump.txt', 'rb') >>> d = pickle.load(f) >>> f.close() >>> d {'age': 20, 'score': 88, 'name': ‘Bob’}

JSON

我們要在不同的編程語言之間傳遞對(duì)象，就必須把對(duì)象序列化為標(biāo)準(zhǔn)格式，比如XML，但更好的方法是序列化為JSON，因?yàn)镴SON表示出來就是一個(gè)字符串，可以被所有語言讀取，也可以方便地存儲(chǔ)到磁盤或者通過網(wǎng)絡(luò)傳輸。JSON不僅是標(biāo)準(zhǔn)格式，并且比XML更快，而且可以直接在Web頁面中讀取，非常方便。

JSON表示的對(duì)象就是標(biāo)準(zhǔn)的JavaScript語言的對(duì)象，JSON和Python內(nèi)置的數(shù)據(jù)類型對(duì)應(yīng)如下:

JSON類型Python類型

{}	dict
[]	list
“string”	str
1234.56	Int或Float
true/false	True/False
null	None

Python內(nèi)置的json模塊提供了非常完善的Python對(duì)象到JSON格式的轉(zhuǎn)換。

>>> json_str = '{"age": 20, "score": 88, "name": "Bob"}' >>> json.loads(json_str) {'age': 20, 'score': 88, 'name': ‘Bob’}

進(jìn)程和線程

進(jìn)程是程序運(yùn)行的最小單位，線程是進(jìn)程內(nèi)部的子任務(wù)。多任務(wù)的實(shí)現(xiàn)模式：多進(jìn)程、多線程、多進(jìn)程＋多線程。

多進(jìn)程

Unix/Linux操作系統(tǒng)提供了一個(gè)fork()系統(tǒng)調(diào)用，它非常特殊。普通的函數(shù)調(diào)用，調(diào)用一次，返回一次，但是fork()調(diào)用一次，返回兩次，因?yàn)椴僮飨到y(tǒng)自動(dòng)把當(dāng)前進(jìn)程（稱為父進(jìn)程）復(fù)制了一份（稱為子進(jìn)程），然后，分別在父進(jìn)程和子進(jìn)程內(nèi)返回。子進(jìn)程永遠(yuǎn)返回0，而父進(jìn)程返回子進(jìn)程的ID。這樣做的理由是，一個(gè)父進(jìn)程可以fork出很多子進(jìn)程，所以，父進(jìn)程要記下每個(gè)子進(jìn)程的ID，而子進(jìn)程只需要調(diào)用getppid()就可以拿到父進(jìn)程的ID。

import osprint('Process (%s) start...' % os.getpid()) # Only works on Unix/Linux/Mac: pid = os.fork() if pid == 0:print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid())) else:print('I (%s) just created a child process (%s).' % (os.getpid(), pid))

由于Windows環(huán)境沒有fork調(diào)用，為了編寫具備跨平臺(tái)能力的代碼，建議使用Python提供的multiprocessing模塊。

multiprocessing

multiprocessing模塊提供了一個(gè)Process類來代表一個(gè)進(jìn)程對(duì)象。創(chuàng)建子進(jìn)程時(shí)，只需要傳入一個(gè)執(zhí)行函數(shù)和函數(shù)的參數(shù)，創(chuàng)建一個(gè)Process實(shí)例，用start()方法啟動(dòng)，這樣創(chuàng)建進(jìn)程比fork()還要簡單。join()方法可以等待子進(jìn)程結(jié)束后再繼續(xù)往下運(yùn)行，通常用于進(jìn)程間的同步。

from multiprocessing import Process import os# 子進(jìn)程要執(zhí)行的代碼 def run_proc(name):print('Run child process %s (%s)...' % (name, os.getpid()))if __name__=='__main__':print('Parent process %s.' % os.getpid())p = Process(target=run_proc, args=('test',))print('Child process will start.')p.start()p.join()print('Child process end.’)

Pool

可以使用進(jìn)程池的方式，創(chuàng)建大量的子進(jìn)程。對(duì)Pool對(duì)象調(diào)用join()方法會(huì)等待所有子進(jìn)程執(zhí)行完畢，調(diào)用join()之前必須先調(diào)用close()，調(diào)用close()之后就不能繼續(xù)添加新的Process了。

from multiprocessing import Pool import os, time, randomdef long_time_task(name):print('Run task %s (%s)...' % (name, os.getpid()))start = time.time()time.sleep(random.random() * 3)end = time.time()print('Task %s runs %0.2f seconds.' % (name, (end - start)))if __name__=='__main__':print('Parent process %s.' % os.getpid())p = Pool(4)for i in range(5):p.apply_async(long_time_task, args=(i,))print('Waiting for all subprocesses done...')p.close()p.join()print('All subprocesses done.')

進(jìn)程間通信

Python的multiprocessing模塊包裝了底層的機(jī)制，提供了Queue、Pipes等多種方式來交換數(shù)據(jù)。

from multiprocessing import Process, Queue import os, time, random# 寫數(shù)據(jù)進(jìn)程執(zhí)行的代碼: def write(q):print('Process to write: %s' % os.getpid())for value in ['A', 'B', 'C']:print('Put %s to queue...' % value)q.put(value)time.sleep(random.random())# 讀數(shù)據(jù)進(jìn)程執(zhí)行的代碼: def read(q):print('Process to read: %s' % os.getpid())while True:value = q.get(True)print('Get %s from queue.' % value)if __name__=='__main__':# 父進(jìn)程創(chuàng)建Queue，并傳給各個(gè)子進(jìn)程：q = Queue()pw = Process(target=write, args=(q,))pr = Process(target=read, args=(q,))# 啟動(dòng)子進(jìn)程pw，寫入:pw.start()# 啟動(dòng)子進(jìn)程pr，讀取:pr.start()# 等待pw結(jié)束:pw.join()# pr進(jìn)程里是死循環(huán)，無法等待其結(jié)束，只能強(qiáng)行終止:pr.terminate()

多線程

多任務(wù)可以由多進(jìn)程完成，也可以由一個(gè)進(jìn)程內(nèi)的多線程完成。一個(gè)進(jìn)程至少有一個(gè)線程。由于線程是操作系統(tǒng)直接支持的執(zhí)行單元，因此，高級(jí)語言通常都內(nèi)置多線程的支持，Python也不例外，并且，Python的線程是真正的Posix Thread，而不是模擬出來的線程。

Python的標(biāo)準(zhǔn)庫提供了兩個(gè)模塊：_thread和threading，_thread是低級(jí)模塊，threading是高級(jí)模塊，對(duì)_thread進(jìn)行了封裝。絕大多數(shù)情況下，我們只需要使用threading這個(gè)高級(jí)模塊。

由于任何進(jìn)程默認(rèn)就會(huì)啟動(dòng)一個(gè)線程，我們把該線程稱為主線程，主線程又可以啟動(dòng)新的線程，Python的threading模塊有個(gè)current_thread()函數(shù)，它永遠(yuǎn)返回當(dāng)前線程的實(shí)例。主線程實(shí)例的名字叫MainThread，子線程的名字在創(chuàng)建時(shí)指定，如果不起名字Python就自動(dòng)給線程命名為Thread-1，Thread-2……

Lock

多線程和多進(jìn)程最大的不同在于，多進(jìn)程中，同一個(gè)變量，各自有一份拷貝存在于每個(gè)進(jìn)程中，互不影響，而多線程中，所有變量都由所有線程共享，所以，任何一個(gè)變量都可以被任何一個(gè)線程修改，因此，線程之間共享數(shù)據(jù)最大的危險(xiǎn)在于多個(gè)線程同時(shí)改一個(gè)變量，把內(nèi)容給改亂了。

要解決上述問題，需要通過加鎖來解決。創(chuàng)建一個(gè)鎖就是通過｀``threading.Lock()```來實(shí)現(xiàn)，當(dāng)多個(gè)線程同時(shí)執(zhí)行l(wèi)ock.acquire()時(shí)，只有一個(gè)線程能成功地獲取鎖，然后繼續(xù)執(zhí)行代碼，其他線程就繼續(xù)等待直到獲得鎖為止。

鎖的好處就是確保了某段關(guān)鍵代碼只能由一個(gè)線程從頭到尾完整地執(zhí)行，壞處當(dāng)然也很多，首先是阻止了多線程并發(fā)執(zhí)行，包含鎖的某段代碼實(shí)際上只能以單線程模式執(zhí)行，效率就大大地下降了。其次，由于可以存在多個(gè)鎖，不同的線程持有不同的鎖，并試圖獲取對(duì)方持有的鎖時(shí)，可能會(huì)造成死鎖，導(dǎo)致多個(gè)線程全部掛起，既不能執(zhí)行，也無法結(jié)束，只能靠操作系統(tǒng)強(qiáng)制終止。

import time, threading# 假定這是你的銀行存款: balance = 0 lock = threading.Lock()def run_thread(n):for i in range(100000):# 先要獲取鎖:lock.acquire()try:# 放心地改吧:change_it(n)finally:# 改完了一定要釋放鎖:lock.release()def run_thread(n):for i in range(100000):change_it(n)t1 = threading.Thread(target=run_thread, args=(5,)) t2 = threading.Thread(target=run_thread, args=(8,)) t1.start() t2.start() t1.join() t2.join() print(balance)

多核CPU

Python的線程雖然是真正的線程，但解釋器執(zhí)行代碼時(shí)，有一個(gè)GIL鎖：Global Interpreter Lock，任何Python線程執(zhí)行前，必須先獲得GIL鎖，然后，每執(zhí)行100條字節(jié)碼，解釋器就自動(dòng)釋放GIL鎖，讓別的線程有機(jī)會(huì)執(zhí)行。這個(gè)GIL全局鎖實(shí)際上把所有線程的執(zhí)行代碼都給上了鎖，所以，多線程在Python中只能交替執(zhí)行，即使100個(gè)線程跑在100核CPU上，也只能用到1個(gè)核。

GIL是Python解釋器設(shè)計(jì)的歷史遺留問題，通常我們用的解釋器是官方實(shí)現(xiàn)的CPython，要真正利用多核，除非重寫一個(gè)不帶GIL的解釋器。

ThreadLocal

在多線程環(huán)境中，每個(gè)線程處理數(shù)據(jù)最好使用局部變量，但是需要在不同線程間傳遞參數(shù)的時(shí)候，會(huì)變的很麻煩。ThreadLocal提供了創(chuàng)建與線程名稱關(guān)聯(lián)的局部變量功能能。ThreadLocal最常用的地方就是為每個(gè)線程綁定一個(gè)數(shù)據(jù)庫連接，HTTP請求，用戶身份信息等，這樣一個(gè)線程的所有調(diào)用到的處理函數(shù)都可以非常方便地訪問這些資源。

import threading# 創(chuàng)建全局ThreadLocal對(duì)象: local_school = threading.local()def process_student():# 獲取當(dāng)前線程關(guān)聯(lián)的student:std = local_school.studentprint('Hello, %s (in %s)' % (std, threading.current_thread().name))def process_thread(name):# 綁定ThreadLocal的student:local_school.student = nameprocess_student()t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A') t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B') t1.start() t2.start() t1.join() t2.join()

進(jìn)程 VS 線程

要實(shí)現(xiàn)多任務(wù)，通常我們會(huì)設(shè)計(jì)Master-Worker模式，Master負(fù)責(zé)分配任務(wù)，Worker負(fù)責(zé)執(zhí)行任務(wù)，因此，多任務(wù)環(huán)境下，通常是一個(gè)Master，多個(gè)Worker。

如果用多進(jìn)程實(shí)現(xiàn)Master-Worker，主進(jìn)程就是Master，其他進(jìn)程就是Worker。

如果用多線程實(shí)現(xiàn)Master-Worker，主線程就是Master，其他線程就是Worker。

多進(jìn)程模式最大的優(yōu)點(diǎn)就是穩(wěn)定性高，因?yàn)橐粋€(gè)子進(jìn)程崩潰了，不會(huì)影響主進(jìn)程和其他子進(jìn)程。（當(dāng)然主進(jìn)程掛了所有進(jìn)程就全掛了，但是Master進(jìn)程只負(fù)責(zé)分配任務(wù)，掛掉的概率低）著名的Apache最早就是采用多進(jìn)程模式。

多進(jìn)程模式的缺點(diǎn)是創(chuàng)建進(jìn)程的代價(jià)大，在Unix/Linux系統(tǒng)下，用fork調(diào)用還行，在Windows下創(chuàng)建進(jìn)程開銷巨大。另外，操作系統(tǒng)能同時(shí)運(yùn)行的進(jìn)程數(shù)也是有限的，在內(nèi)存和CPU的限制下，如果有幾千個(gè)進(jìn)程同時(shí)運(yùn)行，操作系統(tǒng)連調(diào)度都會(huì)成問題。

多線程模式通常比多進(jìn)程快一點(diǎn)，但是也快不到哪去，而且，多線程模式致命的缺點(diǎn)就是任何一個(gè)線程掛掉都可能直接造成整個(gè)進(jìn)程崩潰，因?yàn)樗芯€程共享進(jìn)程的內(nèi)存。在Windows上，如果一個(gè)線程執(zhí)行的代碼出了問題，你經(jīng)常可以看到這樣的提示：“該程序執(zhí)行了非法操作，即將關(guān)閉”，其實(shí)往往是某個(gè)線程出了問題，但是操作系統(tǒng)會(huì)強(qiáng)制結(jié)束整個(gè)進(jìn)程。

計(jì)算密集型 vs. IO密集型

是否采用多任務(wù)的第二個(gè)考慮是任務(wù)的類型。我們可以把任務(wù)分為計(jì)算密集型和IO密集型。

計(jì)算密集型任務(wù)的特點(diǎn)是要進(jìn)行大量的計(jì)算，消耗CPU資源，比如計(jì)算圓周率、對(duì)視頻進(jìn)行高清解碼等等，全靠CPU的運(yùn)算能力。這種計(jì)算密集型任務(wù)雖然也可以用多任務(wù)完成，但是任務(wù)越多，花在任務(wù)切換的時(shí)間就越多，CPU執(zhí)行任務(wù)的效率就越低，所以，要最高效地利用CPU，計(jì)算密集型任務(wù)同時(shí)進(jìn)行的數(shù)量應(yīng)當(dāng)?shù)扔贑PU的核心數(shù)。

計(jì)算密集型任務(wù)由于主要消耗CPU資源，因此，代碼運(yùn)行效率至關(guān)重要。Python這樣的腳本語言運(yùn)行效率很低，完全不適合計(jì)算密集型任務(wù)。對(duì)于計(jì)算密集型任務(wù)，最好用C語言編寫。

第二種任務(wù)的類型是IO密集型，涉及到網(wǎng)絡(luò)、磁盤IO的任務(wù)都是IO密集型任務(wù)，這類任務(wù)的特點(diǎn)是CPU消耗很少，任務(wù)的大部分時(shí)間都在等待IO操作完成（因?yàn)镮O的速度遠(yuǎn)遠(yuǎn)低于CPU和內(nèi)存的速度）。對(duì)于IO密集型任務(wù)，任務(wù)越多，CPU效率越高，但也有一個(gè)限度。常見的大部分任務(wù)都是IO密集型任務(wù)，比如Web應(yīng)用。

IO密集型任務(wù)執(zhí)行期間，99%的時(shí)間都花在IO上，花在CPU上的時(shí)間很少，因此，用運(yùn)行速度極快的C語言替換用Python這樣運(yùn)行速度極低的腳本語言，完全無法提升運(yùn)行效率。對(duì)于IO密集型任務(wù)，最合適的語言就是開發(fā)效率最高（代碼量最少）的語言，腳本語言是首選，C語言最差。

異步IO

考慮到CPU和IO之間巨大的速度差異，一個(gè)任務(wù)在執(zhí)行的過程中大部分時(shí)間都在等待IO操作，單進(jìn)程單線程模型會(huì)導(dǎo)致別的任務(wù)無法并行執(zhí)行，因此，我們才需要多進(jìn)程模型或者多線程模型來支持多任務(wù)并發(fā)執(zhí)行。

現(xiàn)代操作系統(tǒng)對(duì)IO操作已經(jīng)做了巨大的改進(jìn)，最大的特點(diǎn)就是支持異步IO。如果充分利用操作系統(tǒng)提供的異步IO支持，就可以用單進(jìn)程單線程模型來執(zhí)行多任務(wù)，這種全新的模型稱為事件驅(qū)動(dòng)模型，Nginx就是支持異步IO的Web服務(wù)器，它在單核CPU上采用單進(jìn)程模型就可以高效地支持多任務(wù)。在多核CPU上，可以運(yùn)行多個(gè)進(jìn)程（數(shù)量與CPU核心數(shù)相同），充分利用多核CPU。由于系統(tǒng)總的進(jìn)程數(shù)量十分有限，因此操作系統(tǒng)調(diào)度非常高效。用異步IO編程模型來實(shí)現(xiàn)多任務(wù)是一個(gè)主要的趨勢。

對(duì)應(yīng)到Python語言，單進(jìn)程的異步編程模型稱為協(xié)程。

分布式進(jìn)程

Python的multiprocessing模塊不但支持多進(jìn)程，其中managers子模塊還支持把多進(jìn)程分布到多臺(tái)機(jī)器上。一個(gè)服務(wù)進(jìn)程可以作為調(diào)度者，將任務(wù)分布到其他多個(gè)進(jìn)程中，依靠網(wǎng)絡(luò)通信。由于managers模塊封裝很好，不必了解網(wǎng)絡(luò)通信的細(xì)節(jié)，就可以很容易地編寫分布式多進(jìn)程程序。

# task_master.pyimport random, time, queue from multiprocessing.managers import BaseManager# 發(fā)送任務(wù)的隊(duì)列: task_queue = queue.Queue() # 接收結(jié)果的隊(duì)列: result_queue = queue.Queue()# 從BaseManager繼承的QueueManager: class QueueManager(BaseManager):pass# 把兩個(gè)Queue都注冊到網(wǎng)絡(luò)上, callable參數(shù)關(guān)聯(lián)了Queue對(duì)象: QueueManager.register('get_task_queue', callable=lambda: task_queue) QueueManager.register('get_result_queue', callable=lambda: result_queue) # 綁定端口5000, 設(shè)置驗(yàn)證碼'abc': manager = QueueManager(address=('', 5000), authkey=b'abc') # 啟動(dòng)Queue: manager.start() # 獲得通過網(wǎng)絡(luò)訪問的Queue對(duì)象: task = manager.get_task_queue() result = manager.get_result_queue() # 放幾個(gè)任務(wù)進(jìn)去: for i in range(10):n = random.randint(0, 10000)print('Put task %d...' % n)task.put(n) # 從result隊(duì)列讀取結(jié)果: print('Try get results...') for i in range(10):r = result.get(timeout=10)print('Result: %s' % r) # 關(guān)閉: manager.shutdown() print('master exit.’) # task_worker.pyimport time, sys, queue from multiprocessing.managers import BaseManager# 創(chuàng)建類似的QueueManager: class QueueManager(BaseManager):pass# 由于這個(gè)QueueManager只從網(wǎng)絡(luò)上獲取Queue，所以注冊時(shí)只提供名字: QueueManager.register('get_task_queue') QueueManager.register('get_result_queue')# 連接到服務(wù)器，也就是運(yùn)行task_master.py的機(jī)器: server_addr = '127.0.0.1' print('Connect to server %s...' % server_addr) # 端口和驗(yàn)證碼注意保持與task_master.py設(shè)置的完全一致: m = QueueManager(address=(server_addr, 5000), authkey=b'abc') # 從網(wǎng)絡(luò)連接: m.connect() # 獲取Queue的對(duì)象: task = m.get_task_queue() result = m.get_result_queue() # 從task隊(duì)列取任務(wù),并把結(jié)果寫入result隊(duì)列: for i in range(10):try:n = task.get(timeout=1)print('run task %d * %d...' % (n, n))r = '%d * %d = %d' % (n, n, n*n)time.sleep(1)result.put(r)except Queue.Empty:print('task queue is empty.') # 處理結(jié)束: print('worker exit.’)

總結(jié)

以上是生活随笔為你收集整理的Python3.6学习笔记（四）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。