第11章 網絡編程

網絡編程是現代軟件開發中不可或缺的技能,Python提供了豐富的網絡編程庫和工具,讓開發者能夠輕鬆構建各種網絡應用。本章將從基礎的Socket編程開始,逐步深入到HTTP客戶端和服務器開發、異步網絡編程以及網絡安全等高級主題。

本系列文章所使用到的示例源碼:Python從入門到精通示例代碼

11.1 網絡編程基礎

網絡協議概述

網絡協議是計算機網絡中進行數據交換而建立的規則、標準或約定。理解這些協議對於網絡編程至關重要。

TCP/IP協議棧

TCP/IP協議棧是互聯網的基礎,它包含四個層次:

  1. 應用層:HTTP、HTTPS、FTP、SMTP、DNS等
  2. 傳輸層:TCP、UDP
  3. 網絡層:IP、ICMP、ARP
  4. 鏈路層:以太網、WiFi等

常用網絡協議

  • HTTP/HTTPS:超文本傳輸協議,Web應用的基礎
  • FTP:文件傳輸協議
  • SMTP:簡單郵件傳輸協議
  • DNS:域名系統
  • SSH:安全外殼協議

客戶端-服務器模型

客戶端-服務器(C/S)架構是網絡應用的基本模式:

  • 客戶端:發起請求的一方
  • 服務器:響應請求的一方
  • 請求-響應模式:客戶端發送請求,服務器處理並返回響應

IP地址和端口

IP地址

  • IPv4:32位地址,如 192.168.1.1
  • IPv6:128位地址,如 2001:db8::1

端口號

端口號用於標識主機上的不同服務:

  • 知名端口(0-1023):HTTP(80)、HTTPS(443)、FTP(21)、SSH(22)
  • 註冊端口(1024-49151):應用程序註冊使用
  • 動態端口(49152-65535):臨時分配使用

11.2 Socket編程

Socket(套接字)是網絡編程的基礎,它提供了進程間通信的接口。

Socket概念

Socket是網絡通信的端點,可以理解爲網絡編程的”插座”。每個Socket都有一個地址,包括IP地址和端口號。

套接字類型

  • TCP Socket(SOCK_STREAM):面向連接,可靠傳輸
  • UDP Socket(SOCK_DGRAM):無連接,不可靠但高效

Python的socket模塊

Python的socket模塊提供了底層網絡接口:

import socket

# 創建TCP socket
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 創建UDP socket
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

地址族

  • AF_INET:IPv4地址族
  • AF_INET6:IPv6地址族
  • AF_UNIX:Unix域套接字

TCP Socket編程實戰

讓我們通過實際的代碼示例來學習TCP Socket編程。

TCP服務器端

首先創建一個TCP服務器,它能夠處理多個客戶端連接:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TCP服務器示例
"""

import socket
import threading
import time

def handle_client(client_socket, client_address):
    """處理客戶端連接"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_address} 已連接")

    try:
        while True:
            # 接收客戶端數據
            data = client_socket.recv(1024)
            if not data:
                break

            message = data.decode('utf-8')
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 收到來自 {client_address} 的消息: {message}")

            # 回覆客戶端
            response = f"服務器收到: {message}"
            client_socket.send(response.encode('utf-8'))

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 處理客戶端 {client_address} 時發生錯誤: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_address} 已斷開連接")

def start_server():
    """啓動TCP服務器"""
    # 創建socket對象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    # 設置socket選項,允許地址重用
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # 綁定地址和端口
    host = 'localhost'
    port = 8888
    server_socket.bind((host, port))

    # 開始監聽
    server_socket.listen(5)
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] TCP服務器啓動,監聽 {host}:{port}")

    try:
        while True:
            # 接受客戶端連接
            client_socket, client_address = server_socket.accept()

            # 爲每個客戶端創建新線程
            client_thread = threading.Thread(
                target=handle_client,
                args=(client_socket, client_address)
            )
            client_thread.daemon = True
            client_thread.start()

    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器正在關閉...")
    finally:
        server_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器已關閉")

if __name__ == '__main__':
    start_server()

TCP客戶端

接下來創建TCP客戶端來連接服務器:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TCP客戶端示例
"""

import socket
import time

def start_client():
    """啓動TCP客戶端"""
    # 創建socket對象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    try:
        # 連接服務器
        host = 'localhost'
        port = 8888
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 正在連接服務器 {host}:{port}...")
        client_socket.connect((host, port))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已連接到服務器")

        # 發送消息
        messages = [
            "Hello, Server!",
            "這是第二條消息",
            "Python網絡編程測試",
            "再見!"
        ]

        for message in messages:
            # 發送數據
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送消息: {message}")
            client_socket.send(message.encode('utf-8'))

            # 接收服務器響應
            response = client_socket.recv(1024)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器響應: {response.decode('utf-8')}")

            time.sleep(1)  # 等待1秒

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端錯誤: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端已斷開連接")

if __name__ == '__main__':
    start_client()

運行TCP示例

首先啓動服務器:

python tcp_server.py

服務器輸出:

[2025-08-15 21:47:55] TCP服務器啓動,監聽 localhost:8888

然後在另一個終端啓動客戶端:

python tcp_client.py

客戶端輸出:

[2025-08-15 21:48:07] 正在連接服務器 localhost:8888...
[2025-08-15 21:48:07] 已連接到服務器
[2025-08-15 21:48:07] 發送消息: Hello, Server!
[2025-08-15 21:48:07] 服務器響應: 服務器收到: Hello, Server!
[2025-08-15 21:48:08] 發送消息: 這是第二條消息
[2025-08-15 21:48:08] 服務器響應: 服務器收到: 這是第二條消息
[2025-08-15 21:48:09] 發送消息: Python網絡編程測試
[2025-08-15 21:48:09] 服務器響應: 服務器收到: Python網絡編程測試
[2025-08-15 21:48:10] 發送消息: 再見!
[2025-08-15 21:48:10] 服務器響應: 服務器收到: 再見!
[2025-08-15 21:48:11] 客戶端已斷開連接

服務器端同時顯示:

[2025-08-15 21:48:07] 客戶端 ('127.0.0.1', 50531) 已連接
[2025-08-15 21:48:07] 收到來自 ('127.0.0.1', 50531) 的消息: Hello, Server!
[2025-08-15 21:48:08] 收到來自 ('127.0.0.1', 50531) 的消息: 這是第二條消息
[2025-08-15 21:48:09] 收到來自 ('127.0.0.1', 50531) 的消息: Python網絡編程測試
[2025-08-15 21:48:10] 收到來自 ('127.0.0.1', 50531) 的消息: 再見!
[2025-08-15 21:48:11] 客戶端 ('127.0.0.1', 50531) 已斷開連接

UDP Socket編程實戰

UDP是無連接的協議,不需要建立連接就可以發送數據,適合對即時性要求高的應用。

UDP服務器端

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
UDP服務器示例
"""

import socket
import time

def start_udp_server():
    """啓動UDP服務器"""
    # 創建UDP socket對象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    # 綁定地址和端口
    host = 'localhost'
    port = 9999
    server_socket.bind((host, port))

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服務器啓動,監聽 {host}:{port}")

    try:
        while True:
            # 接收數據和客戶端地址
            data, client_address = server_socket.recvfrom(1024)
            message = data.decode('utf-8')

            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 收到來自 {client_address} 的消息: {message}")

            # 回覆客戶端
            response = f"UDP服務器收到: {message}"
            server_socket.sendto(response.encode('utf-8'), client_address)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已回覆客戶端 {client_address}")

    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服務器正在關閉...")
    finally:
        server_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP服務器已關閉")

if __name__ == '__main__':
    start_udp_server()

UDP客戶端

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
UDP客戶端示例
"""

import socket
import time

def start_udp_client():
    """啓動UDP客戶端"""
    # 創建UDP socket對象
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    try:
        # 服務器地址
        server_host = 'localhost'
        server_port = 9999
        server_address = (server_host, server_port)

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客戶端啓動,目標服務器 {server_host}:{server_port}")

        # 發送消息列表
        messages = [
            "Hello UDP Server!",
            "這是UDP通信測試",
            "無連接協議演示",
            "UDP消息結束"
        ]

        for message in messages:
            # 發送數據到服務器
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送消息: {message}")
            client_socket.sendto(message.encode('utf-8'), server_address)

            # 接收服務器響應
            response, server_addr = client_socket.recvfrom(1024)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器 {server_addr} 響應: {response.decode('utf-8')}")

            time.sleep(1)  # 等待1秒

    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客戶端錯誤: {e}")
    finally:
        client_socket.close()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] UDP客戶端已關閉")

if __name__ == '__main__':
    start_udp_client()

運行UDP示例

啓動UDP服務器:

python udp_server.py

服務器輸出:

[2025-08-15 21:48:29] UDP服務器啓動,監聽 localhost:9999

啓動UDP客戶端:

python udp_client.py

客戶端輸出:

[2025-08-15 21:48:36] UDP客戶端啓動,目標服務器 localhost:9999
[2025-08-15 21:48:36] 發送消息: Hello UDP Server!
[2025-08-15 21:48:36] 服務器 ('127.0.0.1', 9999) 響應: UDP服務器收到: Hello UDP Server!
[2025-08-15 21:48:37] 發送消息: 這是UDP通信測試
[2025-08-15 21:48:37] 服務器 ('127.0.0.1', 9999) 響應: UDP服務器收到: 這是UDP通信測試
[2025-08-15 21:48:38] 發送消息: 無連接協議演示
[2025-08-15 21:48:38] 服務器 ('127.0.0.1', 9999) 響應: UDP服務器收到: 無連接協議演示
[2025-08-15 21:48:39] 發送消息: UDP消息結束
[2025-08-15 21:48:39] 服務器 ('127.0.0.1', 9999) 響應: UDP服務器收到: UDP消息結束
[2025-08-15 21:48:40] UDP客戶端已關閉

服務器端顯示:

[2025-08-15 21:48:36] 收到來自 ('127.0.0.1', 57540) 的消息: Hello UDP Server!
[2025-08-15 21:48:36] 已回覆客戶端 ('127.0.0.1', 57540)
[2025-08-15 21:48:37] 收到來自 ('127.0.0.1', 57540) 的消息: 這是UDP通信測試
[2025-08-15 21:48:37] 已回覆客戶端 ('127.0.0.1', 57540)
[2025-08-15 21:48:38] 收到來自 ('127.0.0.1', 57540) 的消息: 無連接協議演示
[2025-08-15 21:48:38] 已回覆客戶端 ('127.0.0.1', 57540)
[2025-08-15 21:48:39] 收到來自 ('127.0.0.1', 57540) 的消息: UDP消息結束
[2025-08-15 21:48:39] 已回覆客戶端 ('127.0.0.1', 57540)

TCP vs UDP 對比

特性 TCP UDP
連接性 面向連接 無連接
可靠性 可靠傳輸 不可靠傳輸
速度 較慢 較快
數據完整性 保證 不保證
適用場景 文件傳輸、網頁瀏覽 視頻直播、遊戲

Socket編程最佳實踐

1. 異常處理

try:
    # socket操作
    pass
except socket.error as e:
    print(f"Socket錯誤: {e}")
except Exception as e:
    print(f"其他錯誤: {e}")
finally:
    # 確保關閉socket
    if 'sock' in locals():
        sock.close()

2. 超時設置

# 設置超時時間
sock.settimeout(10.0)  # 10秒超時

3. 地址重用

# 允許地址重用,避免"Address already in use"錯誤
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

4. 緩衝區管理

# 設置接收緩衝區大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536)

# 設置發送緩衝區大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536)

11.3 HTTP客戶端編程

HTTP(超文本傳輸協議)是Web應用的基礎協議。Python提供了多種方式來進行HTTP客戶端編程。

HTTP協議基礎

HTTP請求方法

  • GET:獲取資源
  • POST:提交數據
  • PUT:更新資源
  • DELETE:刪除資源
  • HEAD:獲取響應頭
  • OPTIONS:獲取支持的方法

HTTP狀態碼

  • 2xx:成功(200 OK、201 Created、204 No Content)
  • 3xx:重定向(301 Moved Permanently、302 Found、304 Not Modified)
  • 4xx:客戶端錯誤(400 Bad Request、401 Unauthorized、404 Not Found)
  • 5xx:服務器錯誤(500 Internal Server Error、502 Bad Gateway、503 Service Unavailable)

urllib模塊

urllib是Python標準庫中的HTTP客戶端模塊,無需安裝第三方庫。

urllib.request基本使用

import urllib.request
import urllib.parse
import json

# GET請求
response = urllib.request.urlopen('https://httpbin.org/get')
print(f"狀態碼: {response.getcode()}")
print(f"響應頭: {dict(response.headers)}")
data = response.read().decode('utf-8')
print(f"響應內容: {data}")

# POST請求
post_data = {'key': 'value', 'name': 'Python'}
data = urllib.parse.urlencode(post_data).encode('utf-8')
req = urllib.request.Request('https://httpbin.org/post', data=data)
response = urllib.request.urlopen(req)
print(f"POST響應: {response.read().decode('utf-8')}")

requests庫

requests庫是Python中最流行的HTTP客戶端庫,提供了更簡潔的API。

安裝requests

pip install requests

requests基本使用

讓我們通過實際的代碼示例來學習HTTP客戶端編程:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
HTTP客戶端編程示例
"""

import urllib.request
import urllib.parse
import urllib.error
import requests
import json
import time

def test_urllib():
    """測試urllib模塊"""
    print("=== urllib模塊測試 ===")

    try:
        # GET請求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送urllib GET請求...")
        response = urllib.request.urlopen('https://httpbin.org/get', timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET請求狀態碼: {response.getcode()}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 響應頭Content-Type: {response.headers.get('Content-Type')}")

        data = json.loads(response.read().decode('utf-8'))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端IP: {data.get('origin')}")

    except urllib.error.URLError as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib GET請求失敗: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib GET請求異常: {e}")

    try:
        # POST請求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送urllib POST請求...")
        post_data = {
            'name': 'Python網絡編程',
            'version': '3.9',
            'framework': 'urllib'
        }

        data = urllib.parse.urlencode(post_data).encode('utf-8')
        req = urllib.request.Request('https://httpbin.org/post', data=data)
        req.add_header('Content-Type', 'application/x-www-form-urlencoded')

        response = urllib.request.urlopen(req, timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST請求狀態碼: {response.getcode()}")

        result = json.loads(response.read().decode('utf-8'))
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 提交的表單數據: {result.get('form')}")

    except urllib.error.URLError as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib POST請求失敗: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] urllib POST請求異常: {e}")

def test_requests():
    """測試requests庫"""
    print("\n=== requests庫測試 ===")

    try:
        # GET請求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送requests GET請求...")
        response = requests.get('https://httpbin.org/get', timeout=5)
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET請求狀態碼: {response.status_code}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 響應頭Content-Type: {response.headers.get('Content-Type')}")

        data = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端IP: {data.get('origin')}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] User-Agent: {data.get('headers', {}).get('User-Agent')}")

    except requests.RequestException as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests GET請求失敗: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests GET請求異常: {e}")

    try:
        # POST請求
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送requests POST請求...")
        post_data = {
            'name': 'Python網絡編程',
            'version': '3.9',
            'framework': 'requests',
            'features': ['簡單易用', '功能強大', '社區活躍']
        }

        headers = {
            'User-Agent': 'Python-HTTP-Client/1.0',
            'Content-Type': 'application/json'
        }

        response = requests.post(
            'https://httpbin.org/post',
            json=post_data,
            headers=headers,
            timeout=5
        )

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST請求狀態碼: {response.status_code}")

        result = response.json()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 提交的JSON數據: {result.get('json')}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 請求頭User-Agent: {result.get('headers', {}).get('User-Agent')}")

    except requests.RequestException as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests POST請求失敗: {e}")
    except Exception as e:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] requests POST請求異常: {e}")

if __name__ == '__main__':
    test_urllib()
    test_requests()

運行HTTP客戶端示例

python http_client_demo.py

實際運行輸出:

=== urllib模塊測試 ===
[2025-08-15 21:48:48] 發送urllib GET請求...
[2025-08-15 21:48:48] urllib GET請求失敗: <urlopen error timed out>
[2025-08-15 21:48:53] 發送urllib POST請求...
[2025-08-15 21:48:53] urllib POST請求失敗: <urlopen error timed out>

=== requests庫測試 ===
[2025-08-15 21:48:58] 發送requests GET請求...
[2025-08-15 21:48:59] GET請求狀態碼: 200
[2025-08-15 21:48:59] 響應頭Content-Type: application/json
[2025-08-15 21:48:59] 客戶端IP: 116.21.38.173
[2025-08-15 21:48:59] User-Agent: python-requests/2.32.3
[2025-08-15 21:48:59] 發送requests POST請求...
[2025-08-15 21:49:00] POST請求狀態碼: 200
[2025-08-15 21:49:00] 提交的JSON數據: {'name': 'Python網絡編程', 'version': '3.9', 'framework': 'requests', 'features': ['簡單易用', '功能強大', '社區活躍']}
[2025-08-15 21:49:00] 請求頭User-Agent: Python-HTTP-Client/1.0

從輸出可以看到,requests庫相比urllib更加穩定和易用,成功完成了HTTP請求。

requests高級功能

會話管理

import requests

# 創建會話對象
session = requests.Session()
session.headers.update({'User-Agent': 'My-App/1.0'})

# 使用會話發送請求
response = session.get('https://httpbin.org/get')
print(response.json())

# 會話會自動處理cookies
session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response = session.get('https://httpbin.org/cookies')
print(response.json())

文件上傳

import requests

# 上傳文件
files = {'file': open('example.txt', 'rb')}
response = requests.post('https://httpbin.org/post', files=files)
print(response.json())

超時和重試

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

# 配置重試策略
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS"]
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

# 發送請求(帶超時和重試)
response = session.get('https://httpbin.org/get', timeout=5)

11.4 Web服務器開發

Python提供了多種方式來開發Web服務器,從簡單的靜態文件服務器到複雜的Web應用框架。

HTTP服務器基礎

HTTP服務器的基本工作流程:

  1. 監聽端口:綁定到指定的IP地址和端口
  2. 接受連接:等待客戶端連接
  3. 解析請求:解析HTTP請求報文
  4. 處理請求:根據請求執行相應的業務邏輯
  5. 生成響應:構造HTTP響應報文
  6. 發送響應:將響應發送給客戶端

http.server模塊

Python標準庫提供了http.server模塊來快速創建HTTP服務器。

簡單HTTP服務器

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
簡單HTTP服務器示例
"""

import http.server
import socketserver
import json
import urllib.parse
import time
from datetime import datetime

class CustomHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
    """自定義HTTP請求處理器"""

    def do_GET(self):
        """處理GET請求"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] GET請求: {self.path} 來自 {self.client_address[0]}")

        if self.path == '/':
            self.send_html_response(self.get_home_page())
        elif self.path == '/api/info':
            self.send_json_response(self.get_server_info())
        elif self.path.startswith('/api/time'):
            self.send_json_response(self.get_current_time())
        else:
            self.send_error(404, "頁面未找到")

    def do_POST(self):
        """處理POST請求"""
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] POST請求: {self.path} 來自 {self.client_address[0]}")

        if self.path == '/api/echo':
            # 讀取請求體
            content_length = int(self.headers.get('Content-Length', 0))
            post_data = self.rfile.read(content_length)

            try:
                # 嘗試解析JSON
                data = json.loads(post_data.decode('utf-8'))
                response = {
                    'status': 'success',
                    'message': '數據接收成功',
                    'received_data': data,
                    'timestamp': datetime.now().isoformat()
                }
                self.send_json_response(response)
            except json.JSONDecodeError:
                # 處理表單數據
                form_data = urllib.parse.parse_qs(post_data.decode('utf-8'))
                response = {
                    'status': 'success',
                    'message': '表單數據接收成功',
                    'form_data': form_data,
                    'timestamp': datetime.now().isoformat()
                }
                self.send_json_response(response)
        else:
            self.send_error(404, "API端點未找到")

    def send_html_response(self, html_content):
        """發送HTML響應"""
        self.send_response(200)
        self.send_header('Content-Type', 'text/html; charset=utf-8')
        self.send_header('Server', 'Python-HTTP-Server/1.0')
        self.end_headers()
        self.wfile.write(html_content.encode('utf-8'))

    def send_json_response(self, data):
        """發送JSON響應"""
        json_data = json.dumps(data, ensure_ascii=False, indent=2)
        self.send_response(200)
        self.send_header('Content-Type', 'application/json; charset=utf-8')
        self.send_header('Server', 'Python-HTTP-Server/1.0')
        self.send_header('Access-Control-Allow-Origin', '*')
        self.end_headers()
        self.wfile.write(json_data.encode('utf-8'))

    def get_home_page(self):
        """獲取首頁HTML"""
        return """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Python HTTP服務器</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 40px; }
        .container { max-width: 800px; margin: 0 auto; }
        .api-list { background: #f5f5f5; padding: 20px; border-radius: 5px; }
        .api-item { margin: 10px 0; }
        .method { color: #007bff; font-weight: bold; }
    </style>
</head>
<body>
    <div class="container">
        <h1>Python HTTP服務器演示</h1>
        <p>這是一個使用Python http.server模塊創建的簡單Web服務器。</p>

        <h2>可用的API端點:</h2>
        <div class="api-list">
            <div class="api-item">
                <span class="method">GET</span> /api/info - 獲取服務器信息
            </div>
            <div class="api-item">
                <span class="method">GET</span> /api/time - 獲取當前時間
            </div>
            <div class="api-item">
                <span class="method">POST</span> /api/echo - 回顯提交的數據
            </div>
        </div>

        <h2>測試示例:</h2>
        <pre>
# 獲取服務器信息
curl http://localhost:8000/api/info

# 提交JSON數據
curl -X POST -H "Content-Type: application/json" \
     -d '{"name":"測試","value":123}' \
     http://localhost:8000/api/echo
        </pre>
    </div>
</body>
</html>
        """

    def get_server_info(self):
        """獲取服務器信息"""
        return {
            'server': 'Python HTTP Server',
            'version': '1.0',
            'timestamp': datetime.now().isoformat(),
            'client_ip': self.client_address[0],
            'user_agent': self.headers.get('User-Agent', 'Unknown')
        }

    def get_current_time(self):
        """獲取當前時間"""
        now = datetime.now()
        return {
            'current_time': now.isoformat(),
            'timestamp': now.timestamp(),
            'formatted_time': now.strftime('%Y年%m月%d日 %H:%M:%S'),
            'timezone': str(now.astimezone().tzinfo)
        }

    def log_message(self, format, *args):
        """自定義日誌格式"""
        pass  # 我們已經在do_GET和do_POST中記錄了日誌

def start_server():
    """啓動HTTP服務器"""
    host = 'localhost'
    port = 8000

    with socketserver.TCPServer((host, port), CustomHTTPRequestHandler) as httpd:
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] HTTP服務器啓動,監聽 http://{host}:{port}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 按 Ctrl+C 停止服務器")

        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器正在關閉...")
        finally:
            httpd.server_close()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器已關閉")

if __name__ == '__main__':
    start_server()

運行HTTP服務器示例

啓動服務器:

python simple_http_server.py

服務器輸出:

[2025-08-15 21:49:08] HTTP服務器啓動,監聽 http://localhost:8000
[2025-08-15 21:49:08] 按 Ctrl+C 停止服務器

測試API端點:

curl http://localhost:8000/api/info

響應結果:

{
  "server": "Python HTTP Server",
  "version": "1.0",
  "timestamp": "2025-08-15T21:49:15.123456",
  "client_ip": "127.0.0.1",
  "user_agent": "curl/8.7.1"
}

服務器日誌:

[2025-08-15 21:49:15] GET請求: /api/info 來自 127.0.0.1

WSGI協議

WSGI(Web Server Gateway Interface)是Python Web應用和Web服務器之間的標準接口。

WSGI應用示例

def simple_wsgi_app(environ, start_response):
    """簡單的WSGI應用"""
    status = '200 OK'
    headers = [('Content-Type', 'text/html; charset=utf-8')]
    start_response(status, headers)

    html = f"""
    <html>
    <body>
        <h1>WSGI應用示例</h1>
        <p>請求方法: {environ['REQUEST_METHOD']}</p>
        <p>請求路徑: {environ['PATH_INFO']}</p>
        <p>查詢字符串: {environ['QUERY_STRING']}</p>
    </body>
    </html>
    """

    return [html.encode('utf-8')]

# 使用wsgiref運行WSGI應用
from wsgiref.simple_server import make_server

if __name__ == '__main__':
    server = make_server('localhost', 8080, simple_wsgi_app)
    print("WSGI服務器啓動在 http://localhost:8080")
    server.serve_forever()

11.5 網絡數據處理

在網絡編程中,數據的序列化、壓縮和加密是常見的需求。

數據序列化

JSON序列化

import json

# Python對象轉JSON
data = {
    'name': '張三',
    'age': 25,
    'skills': ['Python', 'JavaScript', 'SQL'],
    'is_active': True
}

# 序列化
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(f"JSON字符串: {json_str}")

# 反序列化
parsed_data = json.loads(json_str)
print(f"解析後的數據: {parsed_data}")

pickle序列化

import pickle

# 序列化Python對象
data = {'name': '測試', 'numbers': [1, 2, 3, 4, 5]}
serialized = pickle.dumps(data)
print(f"序列化後的字節數據長度: {len(serialized)}")

# 反序列化
deserialized = pickle.loads(serialized)
print(f"反序列化後的數據: {deserialized}")

數據壓縮

gzip壓縮

import gzip
import json

# 原始數據
data = {'message': '這是一個測試消息' * 100}
json_data = json.dumps(data, ensure_ascii=False)

print(f"原始數據大小: {len(json_data.encode('utf-8'))} 字節")

# 壓縮數據
compressed = gzip.compress(json_data.encode('utf-8'))
print(f"壓縮後大小: {len(compressed)} 字節")
print(f"壓縮比: {len(compressed) / len(json_data.encode('utf-8')):.2%}")

# 解壓數據
decompressed = gzip.decompress(compressed).decode('utf-8')
restored_data = json.loads(decompressed)
print(f"解壓後數據正確: {restored_data == data}")

數據加密

hashlib哈希

import hashlib

# 計算哈希值
data = "Python網絡編程"

# MD5哈希
md5_hash = hashlib.md5(data.encode('utf-8')).hexdigest()
print(f"MD5: {md5_hash}")

# SHA256哈希
sha256_hash = hashlib.sha256(data.encode('utf-8')).hexdigest()
print(f"SHA256: {sha256_hash}")

11.6 異步網絡編程

異步編程可以顯著提高網絡應用的性能,特別是在處理大量併發連接時。

異步編程概念

同步vs異步

  • 同步編程:代碼按順序執行,每個操作完成後才執行下一個
  • 異步編程:可以在等待I/O操作時執行其他任務

事件循環

事件循環是異步編程的核心,它負責調度和執行異步任務。

asyncio模塊

Python 3.4+引入了asyncio模塊,提供了異步編程的基礎設施。

基本概念

import asyncio

# 協程函數
async def hello_world():
    print("Hello")
    await asyncio.sleep(1)  # 異步等待
    print("World")

# 運行協程
asyncio.run(hello_world())

異步網絡編程實戰

讓我們創建一個完整的異步網絡編程示例:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
異步網絡編程示例
"""

import asyncio
import json
import time
from datetime import datetime

class AsyncTCPServer:
    """異步TCP服務器"""

    def __init__(self, host='localhost', port=9000):
        self.host = host
        self.port = port
        self.clients = set()
        self.client_count = 0

    async def handle_client(self, reader, writer):
        """處理客戶端連接"""
        client_address = writer.get_extra_info('peername')
        self.client_count += 1
        client_id = self.client_count
        self.clients.add(writer)

        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_id} ({client_address}) 已連接")

        try:
            # 發送歡迎消息
            welcome_msg = f"歡迎連接到異步服務器!您是第 {client_id} 個客戶端\n"
            writer.write(welcome_msg.encode('utf-8'))
            await writer.drain()

            while True:
                # 讀取客戶端數據
                data = await reader.read(1024)
                if not data:
                    break

                message = data.decode('utf-8').strip()
                print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_id} 發送: {message}")

                # 處理特殊命令
                if message.lower() == 'time':
                    response = f"當前時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"
                elif message.lower() == 'clients':
                    response = f"當前連接客戶端數: {len(self.clients)}\n"
                elif message.lower().startswith('broadcast:'):
                    broadcast_msg = message[10:].strip()
                    await self.broadcast_message(f"[廣播] 客戶端 {client_id}: {broadcast_msg}\n", exclude=writer)
                    response = "廣播消息已發送\n"
                elif message.lower() == 'quit':
                    response = "再見!\n"
                    writer.write(response.encode('utf-8'))
                    await writer.drain()
                    break
                else:
                    response = f"服務器收到: {message}\n"

                # 發送響應
                writer.write(response.encode('utf-8'))
                await writer.drain()

        except asyncio.CancelledError:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_id} 連接被取消")
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 處理客戶端 {client_id} 時發生錯誤: {e}")
        finally:
            self.clients.discard(writer)
            writer.close()
            await writer.wait_closed()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_id} ({client_address}) 已斷開連接")

    async def broadcast_message(self, message, exclude=None):
        """向所有客戶端廣播消息"""
        if not self.clients:
            return

        tasks = []
        for client in self.clients.copy():
            if client != exclude and not client.is_closing():
                tasks.append(self.send_to_client(client, message))

        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    async def send_to_client(self, writer, message):
        """向單個客戶端發送消息"""
        try:
            writer.write(message.encode('utf-8'))
            await writer.drain()
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送消息到客戶端失敗: {e}")
            self.clients.discard(writer)

    async def start_server(self):
        """啓動異步服務器"""
        server = await asyncio.start_server(
            self.handle_client,
            self.host,
            self.port
        )

        addr = server.sockets[0].getsockname()
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 異步TCP服務器啓動,監聽 {addr[0]}:{addr[1]}")
        print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 支持的命令: time, clients, broadcast:<消息>, quit")

        async with server:
            await server.serve_forever()

class AsyncTCPClient:
    """異步TCP客戶端"""

    def __init__(self, host='localhost', port=9000):
        self.host = host
        self.port = port
        self.reader = None
        self.writer = None

    async def connect(self):
        """連接到服務器"""
        try:
            self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已連接到服務器 {self.host}:{self.port}")
            return True
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 連接服務器失敗: {e}")
            return False

    async def send_message(self, message):
        """發送消息"""
        if not self.writer:
            return None

        try:
            self.writer.write(message.encode('utf-8'))
            await self.writer.drain()

            # 讀取響應
            response = await self.reader.read(1024)
            return response.decode('utf-8').strip()
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送消息失敗: {e}")
            return None

    async def listen_for_messages(self):
        """監聽服務器消息"""
        try:
            while True:
                data = await self.reader.read(1024)
                if not data:
                    break
                message = data.decode('utf-8').strip()
                if message:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器消息: {message}")
        except Exception as e:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 監聽消息時發生錯誤: {e}")

    async def disconnect(self):
        """斷開連接"""
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 已斷開連接")

async def run_server():
    """運行服務器"""
    server = AsyncTCPServer()
    try:
        await server.start_server()
    except KeyboardInterrupt:
        print(f"\n[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器正在關閉...")

async def run_single_client():
    """運行單個客戶端"""
    client = AsyncTCPClient()

    if not await client.connect():
        return

    try:
        # 啓動消息監聽任務
        listen_task = asyncio.create_task(client.listen_for_messages())

        # 發送測試消息
        messages = [
            "Hello, Async Server!",
            "time",
            "clients",
            "broadcast:大家好,我是異步客戶端!",
            "Python異步網絡編程測試",
            "quit"
        ]

        for message in messages:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 發送消息: {message}")
            response = await client.send_message(message)
            if response:
                print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 服務器響應: {response}")
            await asyncio.sleep(1)

        # 取消監聽任務
        listen_task.cancel()
        try:
            await listen_task
        except asyncio.CancelledError:
            pass

    finally:
        await client.disconnect()

async def run_multiple_clients(num_clients=3):
    """運行多個併發客戶端"""
    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 啓動 {num_clients} 個併發客戶端")

    async def client_task(client_id):
        client = AsyncTCPClient()
        if not await client.connect():
            return

        try:
            # 發送一些測試消息
            for i in range(3):
                message = f"客戶端 {client_id} 的消息 {i+1}"
                response = await client.send_message(message)
                if response:
                    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 客戶端 {client_id} 收到響應: {response}")
                await asyncio.sleep(0.5)

            # 發送廣播消息
            broadcast_msg = f"broadcast:來自客戶端 {client_id} 的廣播"
            await client.send_message(broadcast_msg)
            await asyncio.sleep(1)

            # 斷開連接
            await client.send_message("quit")
        finally:
            await client.disconnect()

    # 併發運行多個客戶端
    tasks = [client_task(i+1) for i in range(num_clients)]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    import sys

    if len(sys.argv) > 1:
        if sys.argv[1] == 'server':
            asyncio.run(run_server())
        elif sys.argv[1] == 'client':
            asyncio.run(run_single_client())
        elif sys.argv[1] == 'multi':
            asyncio.run(run_multiple_clients())
        else:
            print("用法: python async_server_demo.py [server|client|multi]")
    else:
        print("用法: python async_server_demo.py [server|client|multi]")
        print("  server - 啓動異步服務器")
        print("  client - 啓動單個客戶端")
        print("  multi  - 啓動多個併發客戶端")

運行異步網絡編程示例

啓動異步服務器:

python async_server_demo.py server

在另一個終端啓動多個併發客戶端:

python async_server_demo.py multi

異步HTTP客戶端

使用aiohttp庫進行異步HTTP請求:

import aiohttp
import asyncio

async def fetch_url(session, url):
    """異步獲取URL內容"""
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"錯誤: {e}"

async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/3'
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)

        for i, result in enumerate(results):
            print(f"URL {i+1} 結果長度: {len(result)}")

# 運行異步HTTP客戶端
asyncio.run(main())

11.7 網絡安全

網絡安全是網絡編程中的重要主題,需要考慮各種安全威脅和防護措施。

常見網絡安全威脅

1. 注入攻擊

  • SQL注入:惡意SQL代碼注入
  • 命令注入:系統命令注入
  • 代碼注入:惡意代碼注入

2. 跨站攻擊

  • XSS(跨站腳本):注入惡意腳本
  • CSRF(跨站請求僞造):僞造用戶請求

3. 拒絕服務攻擊

  • DoS攻擊:使服務不可用
  • DDoS攻擊:分佈式拒絕服務

安全編程實踐

1. 輸入驗證

import re

def validate_email(email):
    """驗證郵箱格式"""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return re.match(pattern, email) is not None

def sanitize_input(user_input):
    """清理用戶輸入"""
    # 移除危險字符
    dangerous_chars = ['<', '>', '"', "'", '&', ';']
    for char in dangerous_chars:
        user_input = user_input.replace(char, '')
    return user_input.strip()

2. 密碼安全

import hashlib
import secrets

def generate_salt():
    """生成隨機鹽值"""
    return secrets.token_hex(16)

def hash_password(password, salt):
    """哈希密碼"""
    return hashlib.pbkdf2_hmac('sha256', 
                              password.encode('utf-8'), 
                              salt.encode('utf-8'), 
                              100000)

def verify_password(password, salt, hashed):
    """驗證密碼"""
    return hash_password(password, salt) == hashed

SSL/TLS編程

Python的ssl模塊提供了SSL/TLS支持:

import ssl
import socket

# 創建SSL上下文
context = ssl.create_default_context()

# 創建安全連接
with socket.create_connection(('www.python.org', 443)) as sock:
    with context.wrap_socket(sock, server_hostname='www.python.org') as ssock:
        print(f"SSL版本: {ssock.version()}")
        print(f"加密套件: {ssock.cipher()}")

        # 發送HTTPS請求
        ssock.send(b"GET / HTTP/1.1\r\nHost: www.python.org\r\n\r\n")
        response = ssock.recv(4096)
        print(response.decode('utf-8')[:200])

認證和授權

JWT令牌示例

import jwt
import datetime

# 生成JWT令牌
def generate_token(user_id, secret_key):
    payload = {
        'user_id': user_id,
        'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24),
        'iat': datetime.datetime.utcnow()
    }
    return jwt.encode(payload, secret_key, algorithm='HS256')

# 驗證JWT令牌
def verify_token(token, secret_key):
    try:
        payload = jwt.decode(token, secret_key, algorithms=['HS256'])
        return payload['user_id']
    except jwt.ExpiredSignatureError:
        return None  # 令牌已過期
    except jwt.InvalidTokenError:
        return None  # 無效令牌

總結

本章詳細介紹了Python網絡編程的各個方面:

  1. 網絡編程基礎:瞭解了TCP/IP協議棧、客戶端-服務器模型和IP地址端口的概念

  2. Socket編程:掌握了TCP和UDP Socket編程的實現方法,包括服務器端和客戶端的開發

  3. HTTP客戶端編程:學習了使用urllib和requests庫進行HTTP請求的方法

  4. Web服務器開發:瞭解瞭如何使用http.server模塊創建簡單的Web服務器

  5. 網絡數據處理:掌握了數據序列化、壓縮和加密的基本技術

  6. 異步網絡編程:學習了使用asyncio進行異步網絡編程,提高應用性能

  7. 網絡安全:瞭解了常見的網絡安全威脅和防護措施

通過本章的學習,你應該能夠:
- 理解網絡編程的基本概念和原理
- 使用Socket進行底層網絡通信
- 開發HTTP客戶端和服務器應用
- 處理網絡數據的序列化和安全問題
- 使用異步編程提高網絡應用性能
- 實施基本的網絡安全措施

網絡編程是一個廣闊的領域,本章提供了堅實的基礎。在實際項目中,你可能還需要學習特定的Web框架(如Django、Flask)、消息隊列、微服務架構等更高級的主題。記住,網絡編程的關鍵是理解協議、處理併發和確保安全性。

小夜