在現代軟體開發中，非同步程式設計對於提升應用程式效能至關重要。Python 提供了多種工具和技術來實作非同步操作，其中 Future 模式和 Active Object 設計模式是兩種常用的方法。Future 模式允許提交任務並立即傳回一個 Future 物件，代表任務的未來結果，而 Active Object 模式則將方法的執行與呼叫解耦，提高系統回應能力。本文將深入探討這兩種模式在 Python 中的應用，並提供實際程式碼範例。
非同步運算與 Future 模式
非同步運算是一種允許程式在執行任務的同時，也能夠進行其他工作的技術。Future 模式是實作非同步運算的一種常見方法，該模式允許程式提交任務並立即取得一個代表該任務未來結果的物件，即 Future 物件。
Python 中的 Future 模式
Python 的concurrent.futures模組提供了一個實作 Future 模式的類別，即Future類別。以下是一個使用concurrent.futures模組的例子：
import concurrent.futures
import random
import time
def async_operation(task_id):
sleep_time = random.uniform(0.1, 0.5)
time.sleep(sleep_time)
return f"Result for task {task_id} after {sleep_time:.2f}s"
def callback(future):
try:
result = future.result()
print("Callback received:", result)
except Exception as e:
print("Callback encountered exception:", e)
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
futures = {}
for i in range(10):
future = executor.submit(async_operation, i)
future.add_done_callback(callback)
futures[future] = i
while futures:
done, _ = concurrent.futures.wait(futures, timeout=0.2, return_when=concurrent.futures.FIRST_COMPLETED)
for future in done:
task_id = futures.pop(future)
try:
result = future.result()
print(f"Task {task_id} completed with result: {result}")
except Exception as e:
print(f"Task {task_id} failed with exception: {e}")
在這個例子中，async_operation函式模擬了一個非同步運算，Future物件被用來代表該運算的未來結果。callback函式被用來處理運算完成的結果。
Future 模式的優點
Future 模式提供了以下優點：
- 非同步運算：Future 模式允許程式提交任務並立即取得一個代表該任務未來結果的物件，即 Future 物件。
- 回撥機制：Future 模式提供了一個回撥機制，允許程式在運算完成時執行特定的動作。
- 多工：Future 模式允許程式提交多個任務並同時執行，提高了程式的效率。
Future 模式的挑戰
Future 模式也有一些挑戰，例如：
- 取消請求：當一個 Future 物件被取消時，需要確保該物件的資源被釋放，以避免資源洩漏。
- 錯誤處理：當一個 Future 物件發生錯誤時，需要確保該錯誤被正確地處理，以避免程式當機。
自訂 Future 類別
為瞭解決 Future 模式的挑戰，可以自訂一個 Future 類別，以提供更多的控制權。以下是一個自訂 Future 類別的例子：
import threading
import time
class CancellableFuture:
def __init__(self):
self._cancelled = False
self._result = None
self._exception = None
self._lock = threading.Lock()
def cancel(self):
with self._lock:
self._cancelled = True
def result(self):
with self._lock:
if self._cancelled:
raise RuntimeError("Future was cancelled")
if self._exception:
raise self._exception
return self._result
def set_result(self, result):
with self._lock:
self._result = result
def set_exception(self, exception):
with self._lock:
self._exception = exception
在這個例子中，CancellableFuture類別提供了一個可取消的 Future 物件，該物件可以被用來代表一個非同步運算的未來結果。該類別提供了一個cancel方法，用來取消該 Future 物件，一個result方法，用來取得該 Future 物件的結果，一個set_result方法，用來設定該 Future 物件的結果，一個set_exception方法，用來設定該 Future 物件的異常。
平行執行任務的執行緒管理
在進行多執行緒程式設計時，管理執行緒的生命週期和狀態變化至關重要。以下是一個使用 Python 的 threading 模組來實作的執行緒管理類別，該類別允許你建立一個可以被取消的執行緒任務。
類別定義
import threading
import time
class CancellableThread:
def __init__(self, func, *args, **kwargs):
"""
初始化一個可取消的執行緒任務。
:param func: 要執行的函式
:param args: 函式的位置引數
:param kwargs: 函式的關鍵字引數
"""
self._cancelled = threading.Event()
self._done = threading.Event()
self._result = None
self._exception = None
self._thread = threading.Thread(target=self._run, args=(func, args, kwargs))
self._thread.start()
def _run(self, func, args, kwargs):
"""
執行執行緒任務的核心邏輯。
:param func: 要執行的函式
:param args: 函式的位置引數
:param kwargs: 函式的關鍵字引數
"""
try:
# 定期檢查取消訊號
for _ in range(10):
if self._cancelled.is_set():
raise RuntimeError("Operation cancelled")
time.sleep(0.1)
self._result = func(*args, **kwargs)
except Exception as e:
self._exception = e
finally:
self._done.set()
def cancel(self):
"""
取消執行緒任務。
"""
self._cancelled.set()
使用範例
def my_function(x, y):
time.sleep(2) # 模擬一些耗時操作
return x + y
# 建立一個可取消的執行緒任務
task = CancellableThread(my_function, 2, 3)
# 取消執行緒任務
task.cancel()
# 等待執行緒任務完成
task._done.wait()
# 檢查取消狀態
if task._cancelled.is_set():
print("Task was cancelled")
else:
print("Task completed with result:", task._result)
內容解密：
CancellableThread類別使用threading.Event來管理執行緒的取消狀態和完成狀態。_run方法是執行緒任務的核心邏輯，定期檢查取消訊號並執行指定的函式。cancel方法設定取消訊號，通知執行緒任務停止執行。- 使用範例展示瞭如何建立一個可取消的執行緒任務，然後取消它，並等待執行緒任務完成。
圖表翻譯：
@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle
title Python非同步運算Future模式與ActiveObject設計模式
package "非同步程式設計" {
package "事件迴圈" {
component [Event Loop] as loop
component [Task 任務] as task
component [Future 物件] as future
}
package "協程模式" {
component [async/await] as async
component [Generator] as gen
component [Callback] as callback
}
package "並行處理" {
component [asyncio] as asyncio
component [aiohttp] as aiohttp
component [ThreadPool] as thread
component [ProcessPool] as process
}
}
loop --> task : 排程執行
task --> future : 等待結果
async --> asyncio : 協程管理
asyncio --> aiohttp : HTTP 非同步
thread --> process : CPU 密集
note right of loop
單執行緒併發
非阻塞 I/O
高效能處理
end note
@enduml
圖表解釋：
- 建立執行緒任務：初始化
CancellableThread類別。 - 啟動執行緒：啟動執行緒任務的執行。
- 執行執行緒任務：執行指定的函式。
- 檢查取消訊號：定期檢查是否有取消訊號。
- 取消執行緒任務：設定取消訊號，停止執行執行緒任務。
- 繼續執行：繼續執行執行緒任務。
- 完成執行緒任務：完成執行緒任務的執行。
- 設定完成狀態：設定完成狀態，通知等待的主執行緒。
5.4 Future Pattern：管理非同步結果
Future Pattern 是一種用於管理非同步結果的設計模式。它允許開發人員在非同步操作完成之前繼續執行其他任務，而不需要等待結果。
Future Pattern 的優點
- 提高系統回應性：透過使用 Future Pattern，系統可以在非同步操作完成之前繼續執行其他任務，從而提高系統的回應性。
- 簡化錯誤處理：Future Pattern 提供了一種簡單的方式來處理非同步操作的錯誤，透過捕捉和轉發異常，可以確保系統的穩定性。
- 支援取消和超時：Future Pattern 允許開發人員取消非同步操作或設定超時時間，從而可以控制非同步操作的執行時間。
Future Pattern 的實作
import threading
import queue
import time
class Future:
def __init__(self):
self._done = threading.Event()
self._exception = None
self._result = None
def result(self, timeout=None):
finished = self._done.wait(timeout)
if not finished:
raise TimeoutError("Future result not available within timeout")
if self._exception:
raise self._exception
return self._result
def cancel(self):
# 取消非同步操作的實作
pass
class CancellableFuture(Future):
def __init__(self, func, *args, **kwargs):
super().__init__()
self._func = func
self._args = args
self._kwargs = kwargs
def run(self):
try:
result = self._func(*self._args, **self._kwargs)
self._result = result
except Exception as e:
self._exception = e
finally:
self._done.set()
# 使用Future Pattern
def long_running_task(x):
# 模擬一個計算密集型任務
return x * x
future = CancellableFuture(long_running_task, 10)
time.sleep(0.3)
# 取消非同步操作
future.cancel()
try:
res = future.result(timeout=2)
print("任務完成，結果：", res)
except Exception as ex:
print("任務終止，異常：", ex)
Future Pattern 的應用場景
- 非同步操作：Future Pattern 適合於管理非同步操作的結果，例如網路請求、資料函式庫查詢等。
- 並發系統：Future Pattern 可以用於並發系統中，管理多個執行緒或程式之間的非同步結果。
- 分散式系統：Future Pattern 可以用於分散式系統中，管理不同節點之間的非同步結果。
5.5 Active Object Pattern：解耦方法執行
Active Object Pattern 是一種用於解耦方法執行的設計模式。它允許開發人員將方法執行與方法呼叫分離，從而提高系統的回應性和並發性。
Active Object Pattern 的優點
- 提高系統回應性：透過使用 Active Object Pattern，系統可以在方法執行之前繼續執行其他任務，從而提高系統的回應性。
- 簡化並發控制：Active Object Pattern 提供了一種簡單的方式來控制方法執行的並發性，透過使用佇列和鎖機制，可以確保方法執行的安全性。
Active Object Pattern 的實作
import threading
import queue
class ActiveObject:
def __init__(self):
self._queue = queue.Queue()
self._thread = threading.Thread(target=self._run)
self._thread.start()
def _run(self):
while True:
request = self._queue.get()
if request is None:
break
request.method(*request.args, **request.kwargs)
def invoke(self, method, *args, **kwargs):
request = Request(method, args, kwargs)
self._queue.put(request)
class Request:
def __init__(self, method, args, kwargs):
self.method = method
self.args = args
self.kwargs = kwargs
# 使用Active Object Pattern
def my_method(x):
# 方法實作
return x * x
active_object = ActiveObject()
active_object.invoke(my_method, 10)
Active Object Pattern 的應用場景
- 並發系統：Active Object Pattern 適合於並發系統中，管理多個執行緒或程式之間的方法執行。
- 分散式系統：Active Object Pattern 可以用於分散式系統中，管理不同節點之間的方法執行。
- 實時系統：Active Object Pattern 可以用於實時系統中，管理方法執行的時序和優先順序。
從底層實作到高階應用的全面檢視顯示，非同步程式設計模型能有效提升系統效能與反應速度，而 Future 模式和 Active Object 模式則為實踐非同步程式設計提供了兩種截然不同的途徑。Future 模式著重於管理非同步操作的結果，讓開發者無需阻塞等待，便能繼續執行其他任務，尤其適用於 I/O 密集型操作。而 Active Object 模式則專注於解耦方法的執行與呼叫，藉由引入訊息佇列和專屬執行緒，有效簡化了並發控制的複雜度，更適合用於 CPU 密集型操作。
分析兩種模式的實作細節，可以發現 Future 模式更為輕量級，易於理解和應用；但對於複雜的非同步操作流程，Active Object 模式能提供更清晰的結構和更強的控制力。技術限制方面，Future 模式在處理大量非同步操作的取消和錯誤處理時，容易變得繁瑣；而 Active Object 模式則需注意佇列大小的控制和執行緒間的同步問題，避免效能瓶頸或死結發生。
隨著非同步程式設計的普及，預期 Future 模式將持續演進，整合更完善的取消和錯誤處理機制，並與回應式程式設計框架更緊密地結合。Active Object 模式則可能朝向更細粒度的任務排程和資源管理發展，以適應更複雜的並發場景。對於追求高效能和高反應速度的系統而言，選擇合適的非同步程式設計模型至關重要。玄貓認為，開發者應根據具體應用場景和需求，權衡 Future 模式和 Active Object 模式的優劣，才能最大程度地發揮非同步程式設計的優勢。