Python 多執行緒 TCP Client 完整實作教學（自動重連 + 封包記錄）

螃蟹_crab

發佈於Python[基礎][應用][相關]

2026/02/26 更新2026/02/26 發佈閱讀 10 分鐘

本篇教學將帶你實作一個：

✅ 多執行緒 TCP Client
✅ 支援自動斷線重連
✅ 非阻塞收發機制
✅ Queue 非同步發送
✅ JSON 封包格式
✅ 自動依照日期分類儲存封包 Log
✅ Windows Port 被占用處理

這是一個適合工業設備通訊（例如 AVI 檢測系統）或客製化 TCP 協議應用的完整架構範例。

一、整體架構設計

我們的程式包含三個主要類別：

Request        → 封包物件
LogHandler     → 訊息回呼機制
TCPClient      → 核心 TCP 通訊引擎

架構流程如下：

外部程式
   ↓
SendData()
   ↓
tx_queue
   ↓
TCP Thread
   ↓
sendall()
   ↓
Server

Server 回傳
   ↓
recv()
   ↓
rx_queue
   ↓
外部程式處理

二、Request 封包設計

class Request:
    def __init__(self, cmd, data):
        self.Cmd = cmd
        self.Data = data

    def to_json(self):
        return json.dumps({"Cmd": self.Cmd, "Data": self.Data}, ensure_ascii=False)

設計重點

將封包格式統一為：

{
    "Cmd": "PUT_AVIRESULT_REQ",
    "Data": {
        "PN": "ABC123",
        "SN": "0001"
    }
}

ensure_ascii=False
→ 支援中文

這樣可以確保整個系統的封包格式一致。

三、LogHandler 訊息回呼系統

class LogHandler:
    def __init__(self):
        self.callbacks = {}

    def register(self, name, callback):
        self.callbacks[name] = callback

    def emit(self, msg):
        for callback in self.callbacks.values():
            callback(msg)

為什麼需要它？

讓 TCPClient 不直接依賴 UI 或 print。

例如在 PyQt 中可以這樣註冊：

client.logHandlers.register("ui", lambda msg: print(msg))

這樣 TCPClient 就變成：

可嵌入 GUI、Console、Server 程式的通用元件

這是一種 解耦設計（Decoupling Design）

四、TCPClient 核心解析

1️⃣ 初始化

class TCPClient:
    def __init__(self):
        self.remote_addr = ("127.0.0.1", 9999)
        self.local_addr = None
        self.sock = None
        self._isconnected = False
        self._running = False
        self.tx_queue = queue.Queue()
        self.rx_queue = queue.Queue()
        self.logHandlers = LogHandler()
        self.log_root = "C:/AVI_Packet"

重點

變數功能tx_queue發送佇列rx_queue接收佇列_running執行緒是否運行_isconnected是否已連線

五、多執行緒設計

def start(self):
    if self._running: return
    self._running = True
    self.thread = threading.Thread(target=self._tcp_thread, daemon=True)
    self.thread.start()

使用：

daemon=True

代表主程式結束時，TCP thread 會自動關閉。

六、自動重連機制（核心重點）

if not self._isconnected:

重連流程：

關閉舊 socket
建立新 socket
設定 SO_REUSEADDR
綁定 local port
設定 timeout
connect
切回 non-blocking

1️⃣ 為什麼要 shutdown？

self.sock.shutdown(socket.SHUT_RDWR)
self.sock.close()

確保：

舊 socket 立即釋放
不會殘留 TIME_WAIT

2️⃣ Windows Port 10048 問題處理

self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

如果 Port 被佔用：

self.sock.bind((self.local_addr[0], 0))

改用系統自動分配 Port。

這是實務上非常重要的穩定性優化。

3️⃣ 為什麼 connect 之後要設為 non-blocking？

self.sock.setblocking(False)

如果不這樣：

recv() 會卡死

七、收發邏輯解析

1️⃣ 發送資料（優先處理）

if not self.tx_queue.empty():

轉換 JSON：

if isinstance(obj, Request):
    json_str = obj.to_json()

然後發送：

self.sock.sendall((json_str.strip() + "\r\n").encode('utf-8'))

為什麼加 \r\n？

因為我們用它當作封包分隔符。

八、接收資料設計

raw_rx = self.sock.recv(4096)

如果：

if not raw_rx:

代表：

Server 已關閉連線

然後拆分：

msgs = raw_rx.decode('utf-8').split("\r\n")

每筆封包丟入：

self.rx_queue.put(m)

九、斷線判斷機制

捕捉：

except (socket.error, ConnectionResetError)

任何錯誤都視為：

斷線

然後：

_isconnected = False

下一輪自動重連。

十、封包自動存檔系統

C:/AVI_Packet/
    2026/
        0226/
            ABC_SN_timestamp.json

建立路徑：

folder_path = os.path.join(self.log_root,
                           now.strftime("%Y"),
                           now.strftime("%m%d"))

根據 Cmd 命名檔案：

Cmd檔名REGISTER_REQREGISTER_時間.jsonPUT_AVIRESULT_REQPN_SN_時間.json

這對於：

產線追蹤
問題追查
資料留存

非常實用。

十一、智慧休眠設計

if not active_work:
    time.sleep(0.01)
time.sleep(0.005)

目的：

降低 CPU 使用率
不影響即時性

這比無限 while True 空轉好很多。

十二、完整使用範例

client = TCPClient()

client.setRemote("127.0.0.1", 9999)

client.logHandlers.register("console", print)

client.start()

req = Request("PUT_AVIRESULT_REQ", {
    "PN": "ABC123",
    "SN": "0001"
})

client.SendData(req)