人工智能之編程進階 Python高級

第八章 網絡併發類模塊

@


目錄

  • 人工智能之編程進階 Python高級
  • 前言
  • 1. TCP 服務端（監聽連接）
  • 2. TCP 客户端
  • 3. UDP 示例（無連接）
  • 🔁 二、多線程（threading）
  • 1. 基本用法
  • 2. 線程安全（加鎖）
  • ⚡ 三、異步 I/O（asyncio）
  • 1. 基本異步函數
  • 2. 異步 HTTP 請求（需第三方庫）
  • 🧒 四、子進程（subprocess）
  • 1. 簡單運行命令
  • 2. 啓動長期運行的進程
  • 3. 與多進程結合（multiprocessing）
  • 🔄 五、進程 vs 線程：本質區別與選型建議
  • 📌 一句話總結：
  • 💡 實際選型建議：
  • 🔍 六、四大方式對比總結
  • ✅ 七、最佳實踐建議
  • 資料關注



前言

本文針對於網絡編程以及網絡編程中的多線程以及異步編程和子進程進行詳細的敍述，並對這幾種方式進行了對比總結。

---# 🌐 一、網絡編程（socket）

Python 使用 socket 模塊進行底層網絡通信，支持 TCP/UDP。

1. TCP 服務端（監聽連接）

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('localhost', 8888))
server.listen(5)  # 最多排隊 5 個連接

print("服務器啓動，等待連接...")

while True:
    client, addr = server.accept()
    print(f"客户端 {addr} 已連接")
    
    data = client.recv(1024).decode('utf-8')
    print("收到:", data)
    
    client.send(b"Hello from server!")
    client.close()

2. TCP 客户端

import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('localhost', 8888))
client.send(b"Hello server!")
response = client.recv(1024)
print("服務器回覆:", response.decode())
client.close()

3. UDP 示例（無連接）

# 服務端
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('localhost', 9999))
data, addr = sock.recvfrom(1024)
sock.sendto(b"Got it!", addr)

# 客户端
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.sendto(b"Hi UDP!", ('localhost', 9999))
data, _ = sock.recvfrom(1024)

✅ 適用場景：自定義協議、學習網絡原理
❌ 不推薦：寫 Web 服務（用 Flask/FastAPI 更高效）

🔁 二、多線程（threading）

用於 I/O 密集型任務（如文件讀寫、網絡請求），但受 GIL 限制，不能真正並行 CPU 計算。

1. 基本用法

import threading
import time

def worker(name):
    print(f"線程 {name} 開始")
    time.sleep(2)
    print(f"線程 {name} 結束")

# 創建並啓動線程
t1 = threading.Thread(target=worker, args=("A",))
t2 = threading.Thread(target=worker, args=("B",))

t1.start()
t2.start()

t1.join()  # 等待線程結束
t2.join()
print("所有線程完成")

2. 線程安全（加鎖）

lock = threading.Lock()
counter = 0

def increment():
    global counter
    for _ in range(100000):
        with lock:  # 或 lock.acquire() / lock.release()
            counter += 1

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()
print(counter)  # 應為 200000

⚠️ 注意：

  • Python 的 GIL（全局解釋器鎖） 使得同一時間只有一個線程執行 Python 字節碼。
  • 所以多線程適合 I/O 阻塞場景（如下載、數據庫查詢），不適合 CPU 密集型任務。

✅ 典型用途：同時處理多個客户端請求、併發爬蟲（配合 requests）

⚡ 三、異步 I/O（asyncio）

Python 的 原生異步框架，使用 async/await 實現單線程高併發，特別適合 大量 I/O 操作（如 HTTP 請求、數據庫查詢）。

1. 基本異步函數

import asyncio

async def say_hello(delay, name):
    await asyncio.sleep(delay)  # 異步等待（不阻塞其他任務）
    print(f"Hello {name}!")

async def main():
    # 併發運行多個協程
    await asyncio.gather(
        say_hello(1, "Alice"),
        say_hello(2, "Bob"),
        say_hello(0.5, "Charlie")
    )

# 啓動事件循環
asyncio.run(main())

2. 異步 HTTP 請求（需第三方庫）

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://httpbin.org/delay/1"] * 5
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print("全部完成！")

asyncio.run(main())

✅ 優勢：

  • 單線程高併發（萬級連接）
  • 資源佔用低
  • 代碼邏輯清晰（避免回調地獄）

❌ 限制：

  • 所有 I/O 操作必須是 異步兼容的（不能直接用 time.sleep()、requests.get()）
  • 不適合 CPU 密集型任務（會阻塞事件循環）

📌 適用場景：高性能 Web 服務（FastAPI、aiohttp）、實時聊天、爬蟲

🧒 四、子進程（subprocess）

用於 啓動和控制外部程序（如 shell 命令、其他語言腳本），可繞過 GIL，實現真正的 CPU 並行。

1. 簡單運行命令

import subprocess

# 運行命令並獲取輸出
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.stdout)

2. 啓動長期運行的進程

proc = subprocess.Popen(
    ["ping", "baidu.com"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# 讀取輸出（非阻塞需配合 select 或線程）
try:
    stdout, stderr = proc.communicate(timeout=10)
    print(stdout)
except subprocess.TimeoutExpired:
    proc.kill()
    print("超時，已終止")

3. 與多進程結合（multiprocessing）

from multiprocessing import Pool
import os

def cpu_task(n):
    return sum(i*i for i in range(n))  # 純 CPU 計算

if __name__ == '__main__':
    with Pool() as pool:  # 自動使用 CPU 核心數
        results = pool.map(cpu_task, [1000000] * 4)
    print(results)

✅ 關鍵點：

  • subprocess 適合調用 外部程序
  • multiprocessing 適合 Python 內部 CPU 並行
  • 兩者都能繞過 GIL，實現真正並行

📌 典型用途：

  • 調用 ffmpeg 處理視頻
  • 執行 shell 腳本
  • 並行處理圖像/數據（用 multiprocessing）

🔄 五、進程 vs 線程：本質區別與選型建議

這是很多初學者容易混淆的核心問題。下面從多個維度對比：

對比項

線程（Thread）

進程（Process）

定義

同一進程內的執行單元

操作系統分配資源的基本單位

內存空間

共享進程內存（變量、堆）

獨立內存空間（不共享）

創建開銷

小（輕量）

大（需複製內存、資源）

通信方式

直接讀寫共享變量（需加鎖）

需通過 IPC（如 Queue、Pipe、共享內存）

穩定性

一個線程崩潰 → 整個進程崩潰

一個進程崩潰 → 不影響其他進程

GIL 影響

受 GIL 限制，Python 中無法並行 CPU

繞過 GIL，真正並行 CPU

適用任務

I/O 密集型（網絡、磁盤）

CPU 密集型（計算、圖像處理）

Python 模塊

threading

multiprocessing / subprocess

📌 一句話總結：

  • 線程 = 共享內存 + 輕量 + 適合等 I/O
  • 進程 = 獨立內存 + 重量 + 適合幹 CPU 活

💡 實際選型建議：

  • 爬取 100 個網頁？ → 用 threading 或 asyncio（都在等網絡響應）
  • 計算 100 萬個數的平方和？ → 用 multiprocessing（真正並行計算）
  • 調用外部程序（如 ffmpeg）？ → 用 subprocess（天然隔離、安全）

⚠️ 注意：
在 Python 中，由於 GIL（全局解釋器鎖） 的存在，多線程無法提升 CPU 密集型任務的性能。這是選擇進程而非線程的關鍵原因。

🔍 六、四大方式對比總結

方式

是否並行

適用任務類型

是否繞過 GIL

典型場景

多線程 (threading)

❌（偽並行）

I/O 密集型

❌

爬蟲、文件讀寫、多客户端

異步 I/O (asyncio)

❌（單線程）

I/O 密集型

❌

高併發 Web 服務、實時通信

子進程 (subprocess)

✅

調用外部程序

✅

執行 shell、調用 C/Java 程序

多進程 (multiprocessing)

✅

CPU 密集型

✅

數據計算、圖像處理、科學計算

💡 一句話選型：

  • 等網絡/磁盤？ → 用 threading 或 asyncio
  • 算數學/處理數據？ → 用 multiprocessing
  • 跑外部命令？ → 用 subprocess

✅ 七、最佳實踐建議

  1. Web 服務優先選 asyncio（如 FastAPI + async/await），性能遠超多線程。
  2. 不要在 asyncio 中混用阻塞代碼（如 time.sleep()、requests），會卡住整個事件循環。
  3. CPU 密集任務別用 threading，改用 multiprocessing。
  4. 子進程記得處理超時和異常，避免殭屍進程。
  5. 線程間共享數據務必加鎖（Lock、Queue）。
  6. 開發階段可用 concurrent.futures 統一接口：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor