第11章 網絡編程¶
網絡編程是現代軟件開發中不可或缺的技能,Python提供了豐富的網絡編程庫和工具,讓開發者能夠輕鬆構建各種網絡應用。本章將從基礎的Socket編程開始,逐步深入到HTTP客戶端和服務器開發、異步網絡編程以及網絡安全等高級主題。
本系列文章所使用到的示例源碼:Python從入門到精通示例代碼
11.1 網絡編程基礎¶
網絡協議概述¶
網絡協議是計算機網絡中進行數據交換而建立的規則、標準或約定。理解這些協議對於網絡編程至關重要。
TCP/IP協議棧¶
TCP/IP協議棧是互聯網的基礎,它包含四個層次:
- 應用層:HTTP、HTTPS、FTP、SMTP、DNS等
- 傳輸層:TCP、UDP
- 網絡層:IP、ICMP、ARP
- 鏈路層:以太網、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服務器的基本工作流程:
- 監聽端口:綁定到指定的IP地址和端口
- 接受連接:等待客戶端連接
- 解析請求:解析HTTP請求報文
- 處理請求:根據請求執行相應的業務邏輯
- 生成響應:構造HTTP響應報文
- 發送響應:將響應發送給客戶端
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網絡編程的各個方面:
-
網絡編程基礎:瞭解了TCP/IP協議棧、客戶端-服務器模型和IP地址端口的概念
-
Socket編程:掌握了TCP和UDP Socket編程的實現方法,包括服務器端和客戶端的開發
-
HTTP客戶端編程:學習了使用urllib和requests庫進行HTTP請求的方法
-
Web服務器開發:瞭解瞭如何使用http.server模塊創建簡單的Web服務器
-
網絡數據處理:掌握了數據序列化、壓縮和加密的基本技術
-
異步網絡編程:學習了使用asyncio進行異步網絡編程,提高應用性能
-
網絡安全:瞭解了常見的網絡安全威脅和防護措施
通過本章的學習,你應該能夠:
- 理解網絡編程的基本概念和原理
- 使用Socket進行底層網絡通信
- 開發HTTP客戶端和服務器應用
- 處理網絡數據的序列化和安全問題
- 使用異步編程提高網絡應用性能
- 實施基本的網絡安全措施
網絡編程是一個廣闊的領域,本章提供了堅實的基礎。在實際項目中,你可能還需要學習特定的Web框架(如Django、Flask)、消息隊列、微服務架構等更高級的主題。記住,網絡編程的關鍵是理解協議、處理併發和確保安全性。