Python 協程提供一種優雅的機制，簡化非同步任務的處理，提升程式碼效率。相較於傳統的執行緒模型，協程更輕量，上下文切換開銷更小，尤其適用於 I/O 密集型任務。本文將探討協程的原理和應用，並分享一些實戰經驗。
協程本質上是一種特殊的生成器，利用 yield 關鍵字實作流程控制。當執行到 yield 時，協程會暫停執行，並將值傳回給呼叫者。下次呼叫時，則從上次暫停的位置繼續執行。這種機制允許在單一執行緒內交錯執行多個任務，避免了執行緒切換的 overhead。除了 next() 函式，send() 方法允許向協程傳遞資料，指定給 yield 表示式，實作更靈活的互動。yield from 則進一步簡化了協程的委派，讓程式碼結構更清晰。
以 Game of Life 為例，可以清楚地看到協程的優勢。透過定義 Query 和 Transition 類別，以及 game_logic 和 step_cell 函式，我們可以模擬細胞的狀態變化。simulate 協程則負責推動整個遊戲的演化，而 live_a_generation 函式則負責協程與外部環境的互動。這種設計有效地分離了遊戲邏輯和執行環境，提升了程式碼的可維護性。
Python 2 中雖然沒有 yield from 和直接的傳回值機制，但可以透過迴圈和自定義異常來實作類別似的功能。對於需要更高效能的場景，可以考慮使用 concurrent.futures 模組，結合多行程技術突破 GIL 的限制。ProcessPoolExecutor 能夠將任務分配到多個 CPU 核心，實作真正的平行處理，大幅提升計算速度。然而，多行程也引入了額外的複雜性，例如資料序列化和行程間通訊，需要謹慎評估其適用性。
Python協程：玄貓如何用平行處理簡化複雜任務
協程（Coroutines）是Python中一種強大的程式設計工具，它能讓你以更簡潔、高效的方式處理非同步任務。身為玄貓，我在開發高併發系統時，經常利用協程來最佳化效能、提升程式碼的可讀性。今天，就讓我來分享協程的奧妙，以及如何在實際專案中運用它們。
協程是什麼？玄貓的理解
協程可以被視為一種輕量級的執行緒，它允許你在單一執行緒中執行多個任務，而無需傳統執行緒的上下文切換開銷。與執行緒不同，協程的切換是由程式碼主動控制的，這使得它們在處理I/O密集型任務時非常高效。
協程的核心概念：生成器與yield
在Python中，協程是根據生成器（Generator）實作的。生成器是一種特殊的函式，它使用yield陳述式來產生一系列的值，而不是一次性傳回所有結果。當生成器遇到yield時，它會暫停執行，並將yield後面的值傳回給呼叫者。下次呼叫生成器時，它會從上次暫停的地方繼續執行。
def my_generator(n):
for i in range(n):
yield i
# 示範如何使用生成器
my_gen = my_generator(5)
print(next(my_gen)) # 輸出: 0
print(next(my_gen)) # 輸出: 1
print(next(my_gen)) # 輸出: 2
內容解密
my_generator(n): 定義一個生成器函式，接受一個引數n。for i in range(n): 迴圈n次。yield i: 每次迴圈產生一個值i，並暫停函式執行。my_gen = my_generator(5): 建立一個生成器物件，傳入引數 5。next(my_gen): 呼叫next()函式來取得生成器的下一個值。每次呼叫next()，生成器會從上次暫停的地方繼續執行，直到遇到下一個yield。
send()方法：與協程互動
除了使用next()方法來推進協程的執行，我們還可以使用send()方法來向協程傳送資料。當協程遇到yield時，它可以接收send()方法傳送的值，並將其作為yield表示式的結果。
def my_coroutine():
x = yield
print("接收到的值:", x)
# 示範如何使用send()方法
coro = my_coroutine()
next(coro) # 啟動協程
coro.send(10) # 傳送值10給協程，輸出: 接收到的值: 10
內容解密
my_coroutine(): 定義一個協程函式。x = yield: 協程暫停執行，等待接收外部傳送的值，並將該值賦給變數x。print("接收到的值:", x): 印出接收到的值。coro = my_coroutine(): 建立一個協程物件。next(coro): 啟動協程，使其執行到第一個yield陳述式。coro.send(10): 傳送值 10 給協程。協程會接收到這個值，並將其賦給變數x，然後繼續執行。
yield from：協程的委派
當我們需要將一個協程的任務委派給另一個協程時，可以使用yield from語法。這能讓我們更輕鬆地組合多個協程，構建複雜的非同步流程。
def sub_coroutine():
yield 1
yield 2
def main_coroutine():
yield from sub_coroutine()
yield 3
# 示範如何使用yield from
coro = main_coroutine()
print(list(coro)) # 輸出: [1, 2, 3]
內容解密
sub_coroutine(): 定義一個子協程，產生值 1 和 2。main_coroutine(): 定義一個主協程。yield from sub_coroutine(): 將子協程的執行權委派給主協程。主協程會產生子協程的所有值，就像這些值是直接由主協程產生的一樣。yield 3: 主協程產生值 3。coro = main_coroutine(): 建立一個協程物件。print(list(coro)): 將協程產生的所有值轉換為列表並印出。
實際應用：Game of Life
讓玄貓用一個經典的例子——Game of Life，來展示協程的威力。Game of Life是一個細胞自動機，其中每個細胞的狀態（存活或死亡）會根據其鄰居的狀態而變化。
ALIVE = '*'
EMPTY = '-'
class Query(object):
def __init__(self, y, x):
self.y = y
self.x = x
def __repr__(self):
return 'Query(%d, %d)' % (self.y, self.x)
class Transition(object):
def __init__(self, y, x, state):
self.y = y
self.x = x
self.state = state
def __repr__(self):
return 'Transition(%d, %d, %r)' % (self.y, self.x, self.state)
def game_logic(state, neighbors):
if state == ALIVE:
if neighbors < 2:
return EMPTY # Live cell dies
elif neighbors > 3:
return EMPTY # Live cell dies
else:
if neighbors == 3:
return ALIVE # Dead cell becomes alive
return state
def step_cell(y, x):
state = yield Query(y, x)
neighbors = 0
for y_off in (-1, 0, 1):
for x_off in (-1, 0, 1):
if y_off == 0 and x_off == 0:
continue
neighbor = yield Query(y + y_off, x + x_off)
if neighbor == ALIVE:
neighbors += 1
next_state = game_logic(state, neighbors)
yield Transition(y, x, next_state)
內容解密
ALIVE和EMPTY: 定義細胞的兩種狀態。Query類別: 代表查詢細胞狀態的請求。Transition類別: 代表細胞狀態變更的請求。game_logic(state, neighbors): 實作 Game of Life 的規則。step_cell(y, x): 協程函式，用於計算特定細胞的下一個狀態。它首先查詢細胞的當前狀態，然後查詢其鄰居的狀態，最後根據 Game of Life 的規則計算出下一個狀態。
TICK = object()
def simulate(height, width):
while True:
for y in range(height):
for x in range(width):
yield from step_cell(y, x)
yield TICK
內容解密
TICK = object(): 定義一個特殊的標記物件，用於表示一個時間步的結束。simulate(height, width): 協程函式，用於模擬整個網格的 Game of Life 演化過程。它會不斷地遍歷網格中的每個細胞，並呼叫step_cell()協程來計算每個細胞的下一個狀態。
這個simulate協程的精妙之處在於，它完全獨立於周圍的環境。我還沒有定義網格在Python物件中如何表示，Query、Transition和TICK值如何在外部處理，以及遊戲如何獲得其初始狀態。但邏輯是清晰的。每個細胞將透過執行step_cell進行轉換。然後遊戲時鐘將滴答作響。只要simulate協程被推進，這將永遠持續下去。
協程的優勢：關注點分離
協程的優勢在於它們能幫助你專注於要完成的邏輯。它們將程式碼的環境指令與執行這些指令的實作分離。這使你能夠看似平行地執行協程。這也允許你隨著時間的推移改進遵循這些指令的實作，而無需更改協程。
協程的實際應用
現在，我想在一個真實的環境中執行simulate。為此，我需要表示網格中每個細胞的狀態。在這裡，我定義一個類別來包含網格：
class Grid(object):
def __init__(self, height, width):
self.height = height
self.width = width
self.rows = []
for _ in range(self.height):
self.rows.append([EMPTY] * self.width)
def __str__(self):
# ...
網格允許你取得和設定任何坐標的值。超出範圍的坐標將會環繞，使網格表現得像無限迴圈空間。
def query(self, y, x):
return self.rows[y % self.height][x % self.width]
def assign(self, y, x, state):
self.rows[y % self.height][x % self.width] = state
最後，我可以定義一個函式，它解釋從simulate及其所有內部協程中產生（yield）的值。這個函式將來自協程的指令轉換為與周圍環境的互動。它將整個細胞網格向前推進一步，然後傳回一個包含下一個狀態的新網格。
def live_a_generation(grid, sim):
progeny = Grid(grid.height, grid.width)
item = next(sim)
while item is not TICK:
if isinstance(item, Query):
state = grid.query(item.y, item.x)
item = sim.send(state)
else: # Must be a Transition
progeny.assign(item.y, item.x, item.state)
item = next(sim)
return progeny
要檢視這個函式的實際效果，我需要建立一個網格並設定其初始狀態。在這裡，我建立一個經典的形狀，稱為滑翔機。
grid = Grid(5, 9)
grid.assign(0, 3, ALIVE)
grid.assign(1, 4, ALIVE)
grid.assign(2, 2, ALIVE)
grid.assign(2, 3, ALIVE)
grid.assign(2, 4, ALIVE)
print(grid)
>>>
–*–—
–-*–-
—***–-
–––
–––
現在我可以一次推進這個網格一代。你可以看到滑翔機如何根據game_logic函式中的簡單規則在網格上向下和向右移動。
class ColumnPrinter(object):
def __init__(self):
self.columns = []
def append(self, data):
self.columns.append(data)
def __str__(self):
row_max = 0
for row in self.columns:
row_max = max(row_max, len(row.splitlines()) + 1)
rows = [''] * row_max
for j in range(row_max):
for i, row in enumerate(self.columns):
if j < len(row.splitlines()):
line = row.splitlines()[j]
else:
line = ''
if i > 0:
rows[j] += ' | '
rows[j] += line
return '\n'.join(rows)
columns = ColumnPrinter()
sim = simulate(grid.height, grid.width)
for i in range(5):
columns.append(str(grid))
grid = live_a_generation(grid, sim)
print(columns)
>>>
0 | 1 | 2 | 3 | 4
–*–— | ––– | ––– | ––– | –––
–-*–- | —*-*–- | –-*–- | –*–— | –-*–-
—***–- | –**–- | —*-*–- | –-**– | –—*–
––– | –*–— | –**–- | –**–- | –***–
––– | ––– | ––– | ––– | –––
這種方法最好的部分是，我可以更改game_logic函式，而無需更新其周圍的程式碼。我可以使用Query、Transition和TICK的現有機制更改規則或增加更大的影響範圍。這展示了協程如何實作關注點分離，這是一個重要的設計原則。
Python 2 的協程
在 Python 2 中，協程的語法糖不如 Python 3 那麼優雅。主要有兩個限制：
-
沒有
yield from表示式：在 Python 2 中，你需要使用額外的迴圈來委派生成器協程。# Python 2 def delegated(): yield 1 yield 2 def composed(): yield 'A' for value in delegated(): # yield from in Python 3 yield value yield 'B' print list(composed()) >>> ['A', 1, 2, 'B'] -
不支援
return陳述式：在 Python 2 中，你需要定義自己的異常型別，並在想要傳回值時引發它。# Python 2 class MyReturn(Exception): def __init__(self, value): self.value = value def delegated(): yield 1 raise MyReturn(2) # return 2 in Python 3 yield 'Not reached' def composed(): try: for value in delegated(): yield value except MyReturn as e: output = e.value yield output * 4 print list(composed()) >>> [1, 8]
玄貓的重點整理
- 協程提供了一種高效的方式來同時執行數萬個函式。
- 在生成器中，
yield表示式的值將是從外部程式碼傳遞給生成器的send方法的值。 - 協程為你提供了一個強大的工具，用於將程式的核心邏輯與其與周圍環境的互動分離。
- Python 2 不支援
yield from或從生成器傳回值。
玄貓對平行處理的建議：concurrent.futures
在編寫Python程式時，你可能會遇到效能瓶頸。即使在最佳化程式碼之後，程式的執行速度可能仍然無法滿足你的需求。在擁有越來越多CPU核心的現代電腦上，一個合理的解決方案是平行處理。如果可以將程式碼的計算分割成獨立的工作單元，並在多個CPU核心上同時執行，那會怎麼樣？
不幸的是，Python的全域直譯器鎖（GIL）阻止了執行緒中的真正平行（參見第37項：使用執行緒進行阻塞I/O，避免用於平行），因此該選項不可用。另一個常見的建議是使用C語言將效能關鍵程式碼重寫為擴充套件模組。C語言讓你更接近底層，並且可以比Python執行得更快，從而消除了平行的需求。C擴充套件還可以啟動在本機平行執行的執行緒，並利用多個CPU核心。Python的C擴充套件API有詳細的檔案，並且是逃生艙的一個不錯的選擇。
但是用C語言重寫你的程式碼的成本很高。在Python中簡短易懂的程式碼在C語言中可能會變得冗長而複雜。這樣的移植需要大量的測試，以確保其功能與原始Python程式碼等效，並且沒有引入任何錯誤。有時這是值得的，這解釋了Python社群中C擴充套件模組的大量生態系統，這些模組加速了諸如文字解析、影像合成和矩陣數學之類別的事情。甚至還有諸如Cython和Numba之類別的開源工具可以簡化到C的轉換。
問題是，將程式的一個部分移到C語言通常是不夠的。最佳化的Python程式通常沒有一個主要的緩慢來源，而是通常有許多重要的貢獻者。為了獲得C語言的底層和執行緒的好處，你需要移植程式的大部分，從而大大增加了測試需求和風險。必須有一種更好的方法來保護你在Python中的投資，以解決困難的計算問題。
透過concurrent.futures內建模組可以輕鬆存取的multiprocessing內建模組可能正是你需要的。它使Python能夠透過將其他直譯器作為子行程執行來平行利用多個CPU核心。這些子行程與主直譯器是分開的，因此它們的全域。
協程是Python中一個強大的工具，可以幫助你編寫更簡潔、更高效、更易於維護的程式碼。無論你是開發Web應用程式、資料處理管道還是平行計算系統，協程都能派上用場。希望這篇文章能幫助你理解協程的概念，並在你的專案中開始使用它們。
Python多行程加速：突破GIL的限制
在Python中，由於全域直譯器鎖（GIL）的存在，多執行緒並不能真正利用多個CPU核心。每個子行程都是獨立的，可以充分利用一個CPU核心。子行程與主行程之間透過連結進行通訊，接收計算指令並傳回結果。
舉例來說，假設你需要使用Python進行計算密集型任務，並希望利用多個CPU核心。這裡，我們以計算兩個數字的最大公約數（GCD）為例，來模擬更複雜的計算，例如使用Navier-Stokes方程式模擬流體動力學。
def gcd(pair):
a, b = pair
low = min(a, b)
for i in range(low, 0, -1):
if a % i == 0 and b % i == 0:
return i
內容解密
上述程式碼定義了一個 gcd 函式，用於計算一對數字的最大公約數。它首先找出兩個數字中較小的一個，然後從該數字開始向下迭代，尋找能同時整除兩個數字的最大公約數。
在沒有平行處理的情況下，串列執行此函式所需的時間會線性增加。
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
numbers = [(1963309, 2265973), (2030677, 3814172),
(1551645, 2229620), (2039045, 2020802)]
start = time.time()
results = list(map(gcd, numbers))
end = time.time()
print('Took %.3f seconds' % (end - start))
內容解密
這段程式碼使用 map 函式串列執行 gcd 函式，計算一組數字對的最大公約數，並測量執行時間。
在多執行緒環境下執行這段程式碼並不能提高速度，因為GIL會阻止Python平行使用多個CPU核心。以下是使用 concurrent.futures 模組及其 ThreadPoolExecutor 類別，以及兩個工作執行緒（以比對電腦上的CPU核心數量）進行相同計算的範例：
start = time.time()
pool = ThreadPoolExecutor(max_workers=2)
results = list(pool.map(gcd, numbers))
end = time.time()
print('Took %.3f seconds' % (end - start))
內容解密
這段程式碼使用 ThreadPoolExecutor 類別，透過多執行緒平行執行 gcd 函式。但由於GIL的限制，執行速度並未提升，反而因為執行緒啟動和通訊的額外負擔而變慢。
更令人驚訝的是，僅僅更改一行程式碼，就能產生奇蹟。如果將 ThreadPoolExecutor 替換為 concurrent.futures 模組中的 ProcessPoolExecutor，一切都會加速。
start = time.time()
pool = ProcessPoolExecutor(max_workers=2) # 唯一的變更
results = list(pool.map(gcd, numbers))
end = time.time()
print('Took %.3f seconds' % (end - start))
內容解密
這段程式碼將 ThreadPoolExecutor 替換為 ProcessPoolExecutor，利用多行程平行執行 gcd 函式。由於每個行程都有獨立的Python直譯器，因此可以繞過GIL的限制，充分利用多個CPU核心，從而顯著提高執行速度。
在雙核心機器上執行，速度明顯加快！這怎麼可能呢？以下是 ProcessPoolExecutor 類別的實際運作方式（透過 multiprocessing 模組提供的底層結構）：
- 從要對映的
numbers輸入資料中取得每個專案。 - 使用
pickle模組將其序列化為二進位制資料。 - 透過本地socket將序列化的資料從主直譯器行程複製到子直譯器行程。
- 接下來，在子行程中使用
pickle將資料反序列化為Python物件。 - 然後，匯入包含
gcd函式的Python模組。 - 與其他子行程平行處理在輸入資料上執行該函式。
- 將結果序列化回位元組。
- 透過socket將這些位元組複製回去。
- 將位元組反序列化回父行程中的Python物件。
- 最後，將多個子行程的結果合併到一個列表中以傳回。
雖然對於程式設計師來說看起來很簡單，但 multiprocessing 模組和 ProcessPoolExecutor 類別做了大量工作，以使平行處理成為可能。在大多數其他語言中，協調兩個執行緒所需的唯一接觸點是單個鎖或原子操作。由於必須在父行程和子行程之間進行所有序列化和反序列化，因此使用 multiprocessing 的額外負擔很高。
這種方案非常適合某些型別的隔離、高效能任務。所謂隔離，玄貓指的是不需要與程式的其他部分分享狀態的函式。所謂高效能，玄貓指的是隻需要在父行程和子行程之間傳輸少量資料即可進行大量計算的情況。最大公約數演算法就是一個例子，但許多其他數學演算法的工作方式類別似。
如果你的計算不具備這些特徵，那麼 multiprocessing 的額外負擔可能會阻止它透過平行處理來加速你的程式。在這種情況下，multiprocessing 提供了更高階的分享記憶體、跨行程鎖、佇列和代理工具。但是所有這些功能都非常複雜。在Python執行緒分享的單個行程的記憶體空間中推理這些工具已經夠困難了。將這種複雜性擴充套件到其他行程並涉及socket，會使理解起來更加困難。
玄貓建議避免使用 multiprocessing 的所有部分，並透過更簡單的 concurrent.futures 模組使用這些功能。你可以先使用 ThreadPoolExecutor 類別在執行緒中執行隔離、高效能的函式。稍後，你可以轉到 ProcessPoolExecutor 以獲得加速。最後，一旦你完全耗盡了其他選項，你可以考慮直接使用 multiprocessing 模組。
玄貓的重點提醒
- 將CPU瓶頸轉移到C擴充套件模組可能是提高效能的有效方法，同時最大程度地提高你在Python程式碼中的投資。但是，這樣做的成本很高，並且可能會引入錯誤。
multiprocessing模組提供了強大的工具，可以透過最少的努力來平行處理某些型別的Python計算。multiprocessing的強大功能最好透過內建的concurrent.futures模組及其簡單的ProcessPoolExecutor類別來存取。- 應避免使用
multiprocessing模組的高階部分，因為它們非常複雜。
玄貓在本文中探討瞭如何利用Python的 multiprocessing 模組來突破GIL的限制，實作真正的平行處理。透過 ProcessPoolExecutor，我們可以將計算密集型任務分配到多個CPU核心，從而顯著提高程式的執行速度。然而，multiprocessing 的使用也帶來了額外的複雜性，需要仔細評估其適用性。
Python 內建模組：玄貓的效率提升秘訣
Python 的標準函式庫向來以「內含電池」著稱。相較於其他語言可能僅提供少數核心套件，Python 在預設安裝中就包含了常用功能所需的模組。雖然 Python 社群也貢獻了大量的模組，但 Python 仍致力於在預設環境中提供最重要的模組，以滿足常見的程式開發需求。
標準模組數量龐大，無法在此一一詳述。然而，有些內建模組與 Python 的慣用寫法緊密相連，幾乎可說是語言規格的一部分。這些模組在編寫複雜與容易出錯的程式碼時尤其重要。
善用 functools.wraps 開發更強大的函式裝飾器
Python 提供了特殊的語法來定義函式裝飾器。裝飾器能在函式呼叫前後執行額外的程式碼，讓開發者能存取、修改輸入引數和回傳值。這項功能在實施語意、除錯、註冊函式等方面非常有用。
舉例來說，假設你想印出函式呼叫的引數和回傳值。這在除錯遞迴函式的呼叫堆積疊時特別有幫助。以下是一個範例裝飾器的定義：
from functools import wraps
def trace(func):
@wraps(func)
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
print(f'{func.__name__}({args!r}, {kwargs!r}) -> {result!r}')
return result
return wrapper
你可以使用 @ 符號將此裝飾器應用於函式。
@trace
def fibonacci(n):
"""回傳第 n 個 Fibonacci 數"""
if n in (0, 1):
return n
return (fibonacci(n - 2) + fibonacci(n - 1))
@ 符號等同於呼叫裝飾器，並將回傳值賦予相同作用域中的原始名稱。
fibonacci = trace(fibonacci)
呼叫這個被裝飾的函式，會在 fibonacci 執行前後執行 wrapper 程式碼，印出遞迴堆積疊中每一層的引數和回傳值。
fibonacci(3)
fibonacci((1,), {}) -> 1
fibonacci((0,), {}) -> 0
fibonacci((1,), {}) -> 1
fibonacci((2,), {}) -> 1
fibonacci((3,), {}) -> 2
這個方法運作良好，但有個副作用。裝飾器回傳的值（也就是上面被呼叫的函式）並不會認為它的名稱是 fibonacci。
print(fibonacci)
<function trace.<locals>.wrapper at 0x107f7ed08>
原因並不難理解。trace 函式回傳了它所定義的 wrapper。由於裝飾器的關係，wrapper 函式被賦予了 fibonacci 這個名稱。這種行為會破壞像是除錯器（參見條款 57：「考慮使用 pdb 進行互動式除錯」）和物件序列化器（參見條款 44：「使用 copyreg 讓 pickle 更可靠」）等需要進行內省的工具。
舉例來說，內建的 help 函式在被裝飾的 fibonacci 函式上就變得毫無用處。
help(fibonacci)
Help on function wrapper in module __main__:
wrapper(*args, **kwargs)
解決方案是使用 functools 內建模組中的 wraps 輔助函式。這是一個能協助你撰寫裝飾器的裝飾器。將它應用於 wrapper 函式，會將內部函式的所有重要 metadata 複製到外部函式。
from functools import wraps
def trace(func):
@wraps(func)
def wrapper(*args, **kwargs):
# ...
return wrapper
return wrapper
@trace
def fibonacci(n):
# ...
現在，即使函式被裝飾了，執行 help 函式也能產生預期的結果。
help(fibonacci)
Help on function fibonacci in module __main__:
fibonacci(n)
回傳第 n 個 Fibonacci 數
呼叫 help 只是裝飾器可能導致問題的一個例子。Python 函式還有許多其他的標準屬性（例如 __name__、__module__），必須保留這些屬性才能維持函式在語言中的介面。使用 wraps 可確保你總是能獲得正確的行為。
重點整理
- 裝飾器是 Python 的一種語法，允許一個函式在執行時修改另一個函式。
- 使用裝飾器可能會導致需要進行內省的工具（例如除錯器）產生奇怪的行為。
- 在定義自己的裝飾器時，使用
functools內建模組中的wraps裝飾器，以避免任何問題。
善用 contextlib 和 with 陳述式實作可重複使用的 try/finally 行為
Python 中的 with 陳述式用於指示程式碼在特殊的上下文中執行。舉例來說，互斥鎖（參見條款 38：「使用 Lock 來防止執行緒中的資料競爭」）可以在 with 陳述式中使用，以指示縮排的程式碼僅在持有鎖定時執行。
from threading import Lock
lock = Lock()
with lock:
print('已持有鎖定')
上面的例子等同於這個 try/finally 結構，因為 Lock 類別正確地啟用了 with 陳述式。
lock.acquire()
try:
print('已持有鎖定')
finally:
lock.release()
with 陳述式的版本更好，因為它消除了編寫重複 try/finally 結構程式碼的需要。你可以使用 contextlib 內建模組，讓你的物件和函式能夠在 with 陳述式中使用。
這個模組包含 contextmanager 裝飾器，能讓簡單的函式在 with 陳述式中使用。這比定義一個具有 __enter__ 和 __exit__ 特殊方法的新類別（標準方法）容易得多。
