第四十七篇：网络编程——TCP/IP 与 Socket
前言
网络编程是 Python 最实用的应用领域之一。无论是 Web 服务器、API 客户端、即时通讯还是文件传输，底层都离不开 Socket 通信。本文将系统讲解 TCP/IP 协议基础、Socket API 用法及实战网络应用。
TCP/IP 协议基础
分层模型概览
┌─────────────────────────────────┐
│ 应用层 HTTP / FTP / SMTP │ ← 你的代码工作在这里
│ ─────────────────────────────── │
│ 传输层 TCP / UDP │ ← Socket API 封装了这里
│ ─────────────────────────────── │
│ 网际层 IP │ ← 路由与寻址
│ ─────────────────────────────── │
│ 网络接口层 Ethernet / Wi-Fi │ ← 物理传输
└─────────────────────────────────┘
TCP vs UDP 对比
| 特性 | TCP | UDP |
|---|---|---|
| 连接方式 | 面向连接（三次握手） | 无连接 |
| 可靠性 | 可靠传输，保证顺序 | 不可靠，可能丢包 |
| 速度 | 较慢（有 overhead） | 较快 |
| 流量控制 | 滑动窗口、拥塞控制 | 无 |
| 适用场景 | HTTP、文件传输、邮件 | 视频直播、游戏、DNS |
| 数据量 | 无限流式 | 单个数据包限制 |
Socket 基础 API
创建第一个 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)
# 查看 Socket 信息
print(f"地址族: {tcp_socket.family}") # AF_INET = IPv4
print(f"Socket 类型: {tcp_socket.type}") # SOCK_STREAM = TCP
# 获取本地主机名和 IP
hostname = socket.gethostname()
ip_address = socket.gethostbyname(hostname)
print(f"主机名: {hostname}")
print(f"IP 地址: {ip_address}")
# 反向查询
reverse_name = socket.getfqdn("8.8.8.8")
print(f"8.8.8.8 的域名: {reverse_name}")
# 常用端口号速查
service_ports = {
"HTTP": 80,
"HTTPS": 443,
"FTP": 21,
"SSH": 22,
"SMTP": 25,
"DNS": 53,
"MySQL": 3306,
"Redis": 6379,
}
for service, port in service_ports.items():
print(f"{service:>8s}: {port}")
TCP 服务器与客户端
基本 TCP 服务器
import socket
import threading
import sys
class TCPServer:
"""简易多线程 TCP 服务器"""
def __init__(self, host='127.0.0.1', port=8888):
self.host = host
self.port = port
self.server_socket = None
self.clients = {}
self.client_counter = 0
def start(self):
"""启动服务器"""
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 允许端口重用
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
self.server_socket.bind((self.host, self.port))
self.server_socket.listen(5)
print(f"✓ 服务器启动于 {self.host}:{self.port}")
while True:
client_socket, client_address = self.server_socket.accept()
self.client_counter += 1
thread_id = threading.current_thread().ident
print(f"[{thread_id}] 新连接: {client_address[0]}:{client_address[1]}")
# 为新客户端启动处理线程
client_thread = threading.Thread(
target=self.handle_client,
args=(client_socket, client_address),
daemon=True
)
client_thread.start()
except KeyboardInterrupt:
print("\n服务器关闭中...")
finally:
self.shutdown()
def handle_client(self, client_socket, client_address):
"""处理单个客户端连接"""
try:
client_socket.send(b"Welcome! Type 'quit' to exit.\r\n")
while True:
data = client_socket.recv(4096)
if not data:
break
message = data.decode('utf-8').strip()
print(f"[{client_address[0]}] {message}")
# 回显消息（ECHO 服务）
response = f"[Echo] You said: {message}"
client_socket.sendall(response.encode('utf-8'))
if message.lower() == 'quit':
break
except ConnectionResetError:
print(f"[{client_address[0]}] 连接已断开")
except Exception as e:
print(f"错误: {e}")
finally:
client_socket.close()
print(f"[{client_address[0]}] 连接已关闭")
def shutdown(self):
if self.server_socket:
self.server_socket.close()
if __name__ == "__main__":
import os
server = TCPServer(port=8888)
print(f"运行 PID: {os.getpid()}")
server.start()
基本 TCP 客户端
import socket
def tcp_client(host='127.0.0.1', port=8888):
"""TCP 客户端"""
try:
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect((host, port))
# 接收欢迎消息
welcome = client.recv(4096).decode('utf-8')
print(welcome.rstrip())
# 交互式发送消息
import sys
while True:
message = input("> ")
if message.lower() in ('quit', 'exit'):
break
client.sendall(message.encode('utf-8'))
response = client.recv(4096).decode('utf-8')
print(response)
client.close()
print("已断开连接")
except ConnectionRefusedError:
print(f"无法连接到 {host}:{port}")
except KeyboardInterrupt:
print("\n客户端退出")
# tcp_client()
带超时控制的服务器
import socket
def robust_server(host='0.0.0.0', port=9999, timeout=30):
"""带超时控制的健壮服务器"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(5)
print(f"✓ 服务器就绪: {host}:{port} (超时={timeout}s)")
try:
while True:
client, addr = server.accept()
client.settimeout(timeout) # 设置读取超时
try:
# 接收命令
command = client.recv(1024).decode('utf-8').strip()
print(f"[{addr}] 收到: {command}")
if command == 'ping':
client.sendall(b'pong')
elif command.startswith('date'):
import datetime
now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
client.sendall(now.encode('utf-8'))
elif command.startswith('whoami'):
import getpass, socket
info = f'{getpass.getuser()}@{socket.gethostname()}'
client.sendall(info.encode('utf-8'))
else:
client.sendall(f'Unknown command: {command}'.encode('utf-8'))
except socket.timeout:
print(f"[{addr}] 超时，断开连接")
client.close()
except ConnectionResetError:
print(f"[{addr}] 连接重置")
except Exception as e:
print(f"[{addr}] 错误: {e}")
except KeyboardInterrupt:
print("服务器停止")
finally:
server.close()
# robust_server()
UDP 编程
UDP 广播发送器
import socket
def udp_broadcaster(target_ip='255.255.255.255', port=9999, message="Hello World"):
"""UDP 广播发送器"""
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
packet = f"[BROADCAST] {message}".encode('utf-8')
sock.sendto(packet, (target_ip, port))
print(f"广播已发送: {packet.decode()}")
# udp_broadcaster()
UDP 回显服务器
import socket
def udp_echo_server(host='127.0.0.1', port=9998):
"""UDP 回显服务器"""
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.bind((host, port))
print(f"UDP 回显服务器就绪于 {host}:{port}")
try:
while True:
data, client_addr = server.recvfrom(4096)
message = data.decode('utf-8')
print(f"[{client_addr}] {message}")
# 立即回传
response = f"UDP Echo: {message.upper()}"
server.sendto(response.encode('utf-8'), client_addr)
except KeyboardInterrupt:
print("UDP 服务器关闭")
finally:
server.close()
# udp_echo_server()
HTTP 协议的原始 Socket 实现
手动实现 HTTP GET
import socket
def http_get_manual(url, path='/'):
"""使用原始 Socket 发送 HTTP 请求"""
from urllib.parse import urlparse
parsed = urlparse(url)
host = parsed.hostname or url.split('/')[2]
port = parsed.port or (443 if 'https' in url else 80)
# 构建 HTTP 请求头
request = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"Connection: close\r\n"
f"\r\n"
).encode('utf-8')
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
sock.connect((host, port))
sock.sendall(request)
response_data = b""
while True:
chunk = sock.recv(4096)
if not chunk:
break
response_data += chunk
sock.close()
# 解析响应
header_end = response_data.find(b'\r\n\r\n')
headers = response_data[:header_end].decode('utf-8')
body = response_data[header_end + 4:]
print("=== HTTP 响应 ===")
print(headers)
print("-" * 40)
print(body.decode('utf-8', errors='ignore')[:500])
return body
except Exception as e:
print(f"HTTP 请求失败: {e}")
return b""
# http_get_manual("http://example.com")
简单的 HTTP 服务器
import socket
from datetime import datetime
def simple_http_server(host='127.0.0.1', port=8080):
"""最简单的 HTTP 服务器（不需要第三方库）"""
html_template = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>Python HTTP Server</title>
<style>
body {{ font-family: sans-serif; max-width: 800px; margin: 40px auto; padding: 0 20px; }}
h1 {{ color: #654ACB; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #654ACB; color: white; }}
</style>
</head>
<body>
<h1>🐍 Python HTTP 服务器</h1>
<p>这是由原始 Socket 实现的 HTTP 服务器。</p>
<h2>服务器信息</h2>
<table>
<tr><td>时间</td><td>{time}</td></tr>
<tr><td>路径</td><td>{path}</td></tr>
<tr><td>User-Agent</td><td>{user_agent}</td></tr>
<tr><td>Method</td><td>{method}</td></tr>
</table>
<h2>已处理的请求数</h2>
<p>{count} 次</p>
<footer style="margin-top: 30px; color: gray;">Powered by Python Socket</footer>
</body>
</html>""".format
request_count = [0]
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((host, port))
server_socket.listen(5)
print(f"HTTP 服务器启动于 http://{host}:{port}/")
try:
while True:
client, addr = server_socket.accept()
try:
request_data = client.recv(8192).decode('utf-8', errors='ignore')
lines = request_data.split('\r\n')
if not lines:
continue
# 解析请求行
request_line = lines[0].split(' ')
method = request_line[0] if len(request_line) > 0 else 'UNKNOWN'
path = request_line[1] if len(request_line) > 1 else '/'
# 提取 User-Agent
user_agent = ''
for line in lines:
if line.lower().startswith('user-agent:'):
user_agent = line.split(':', 1)[1].strip()
# 生成 HTML 响应
content = html_template(
time=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
path=path,
user_agent=user_agent or 'N/A',
method=method,
count=request_count[0]
)
request_count[0] += 1
response = (
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/html; charset=utf-8\r\n"
"Content-Length: {}\r\n".format(len(content.encode('utf-8'))) +
"Connection: close\r\n"
"\r\n"
+ content
)
client.sendall(response.encode('utf-8'))
print(f"[{addr[0]}] {method} {path} → 200 ({len(content)} bytes)")
except Exception as e:
print(f"请求处理错误: {e}")
except KeyboardInterrupt:
print("\nHTTP 服务器已停止")
finally:
server_socket.close()
# simple_http_server(port=8080)
Socket 高级特性
非阻塞 I/O
import socket
import select
import time
class AsyncServer:
"""基于 select 的非阻塞异步服务器"""
def __init__(self, host='0.0.0.0', port=8765):
self.host = host
self.port = port
self.inputs = []
self.outputs = {}
def start(self):
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server.setblocking(False)
self.server.bind((self.host, self.port))
self.server.listen(10)
self.inputs.append(self.server)
print(f"异步服务器启动于 {self.host}:{self.port}")
while self.inputs:
readable, writable, exceptional = select.select(
self.inputs, self.outputs, self.inputs
)
for s in readable:
if s is self.server:
conn, addr = self.server.accept()
conn.setblocking(False)
self.inputs.append(conn)
print(f"新连接: {addr}")
else:
try:
data = s.recv(4096)
if data:
s.setblocking(True)
s.sendall(data.upper())
s.setblocking(False)
else:
self.inputs.remove(s)
s.close()
print(f"连接断开: {s.getpeername()}")
except BlockingIOError:
pass
for s in writable:
if s in self.outputs:
msg = self.outputs.pop(s)
s.sendall(msg)
for s in exceptional:
self.inputs.remove(s)
if s in self.outputs:
del self.outputs[s]
s.close()
print(f"异常断开: {s.getpeername()}")
# AsyncServer().start()
SSL/TLS 加密 Socket
import socket
import ssl
import tempfile
import os
def create_self_signed_cert(cert_file='server.crt', key_file='server.key'):
"""生成自签名证书（仅用于开发测试）"""
import subprocess
try:
subprocess.run([
'openssl', 'req', '-x509', '-newkey', 'rsa:4096',
'-keyout', key_file, '-out', cert_file,
'-days', '365', '-nodes',
'-subj', '/CN=localhost'
], check=True, capture_output=True)
print(f"证书已生成: {cert_file}")
except FileNotFoundError:
print("OpenSSL 未安装，跳过证书生成")
return False
return True
def ssl_test_server(host='127.0.0.1', port=8443):
"""TLS 加密服务器示例"""
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
# 实际使用时需要加载真正的证书
# ctx.load_cert_chain(certfile='server.crt', keyfile='server.key')
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(1)
print(f"等待 TLS 连接...")
wrapped = ctx.wrap_socket(server, server_side=True)
client_sock, addr = wrapped.accept()
client_sock.sendall(b"Secure connection established!\n")
print(f"安全连接来自: {addr}")
client_sock.close()
server.close()
# ssl_test_server()
实用工具函数
端口扫描器
import socket
import threading
import time
def port_scanner(target_host, ports=None, timeout=0.01):
"""快速端口扫描器"""
if ports is None:
ports = list(range(1, 1025))
open_ports = []
lock = threading.Lock()
def scan_port(port):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((target_host, port))
if result == 0:
with lock:
open_ports.append((port, banner_map(port)))
sock.close()
except:
pass
banner_map = {
20: "FTP-DATA", 21: "FTP", 22: "SSH", 23: "Telnet",
25: "SMTP", 53: "DNS", 80: "HTTP", 110: "POP3",
143: "IMAP", 443: "HTTPS", 993: "IMAPS", 995: "POP3S",
3306: "MySQL", 3389: "RDP", 5432: "PostgreSQL",
6379: "Redis", 8080: "HTTP-Proxy", 8443: "HTTPS-Alt",
}
threads = []
for port in ports:
t = threading.Thread(target=scan_port, args=(port,))
threads.append(t)
t.start()
for t in threads:
t.join()
return sorted(open_ports)
def scan_common_ports(target='127.0.0.1'):
"""扫描常见端口"""
common = [21, 22, 23, 25, 53, 80, 110, 143, 443, 993, 995,
3306, 3389, 5432, 6379, 8080, 8443]
results = port_scanner(target, common)
print(f"\n{'端口':>6} {'服务':>15} 状态")
print("=" * 30)
for port, service in results:
status = service or "open"
print(f"{port:>6} {status:>15} OPEN")
if not results:
print("(没有发现开放端口)")
return results
# results = scan_common_ports('127.0.0.1')
简单聊天室
import socket
import threading
import json
import queue
class ChatRoom:
"""在线聊天室服务器"""
def __init__(self, host='127.0.0.1', port=9990):
self.host = host
self.port = port
self.clients = {} # socket -> username
self.message_queue = queue.Queue()
def broadcast(self, message, sender=None):
"""向所有客户端广播消息"""
data = message.encode('utf-8')
dead_clients = []
for client_sock in self.clients:
try:
client_sock.sendall(data)
except:
dead_clients.append(client_sock)
for c in dead_clients:
self.clients.pop(c, None)
def handle_client(self, client_socket, address):
"""处理客户端消息"""
try:
# 等待用户名注册
data = client_socket.recv(4096).decode('utf-8')
username = data.strip()
if not username:
client_socket.close()
return
self.clients[client_socket] = username
print(f"[{address}] {username} 加入聊天室")
self.broadcast(f"[SYSTEM] {username} 加入了房间！\n")
while True:
try:
msg = client_socket.recv(4096).decode('utf-8')
if not msg:
break
full_message = f"[{username}] {msg}"
self.broadcast(full_message)
except Exception:
break
except Exception as e:
print(f"错误: {e}")
finally:
username = self.clients.pop(client_socket, 'unknown')
print(f"[{address}] {username} 离开了聊天室")
self.broadcast(f"[SYSTEM] {username} 离开了房间！")
client_socket.close()
def start(self):
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((self.host, self.port))
server.listen(50)
print(f"✓ 聊天室启动于 {self.host}:{self.port}")
try:
while True:
client, addr = server.accept()
thread = threading.Thread(
target=self.handle_client,
args=(client, addr),
daemon=True
)
thread.start()
except KeyboardInterrupt:
print("\n聊天室关闭")
finally:
server.close()
# chatroom = ChatRoom()
# chatroom.start()
最佳实践总结
✅ 推荐做法
- 始终设置超时 —
settimeout()防止永久阻塞 - 使用 SO_REUSEADDR — 避免重启时端口被占用
- 捕获 ConnectionResetError — 处理意外断开的连接
- 线程池管理连接 — 高并发时避免创建过多线程
- 区分 TCP/UDP — 需要可靠性用 TCP，需要实时性用 UDP
- 验证输入数据 — 防御注入攻击和数据截断
❌ 避免做法
import socket
# ❌ 不设置超时
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("slow-server.example.com", 80)) # 可能永久阻塞!
# ✅ 设置合理超时
sock.settimeout(10)
# ❌ 忽略连接重置异常
try:
data = sock.recv(4096)
except OSError:
pass # 吞掉所有错误，包括 ConnectionResetError!
# ✅ 分别处理不同异常
try:
data = sock.recv(4096)
except ConnectionResetError:
print("对端强制关闭了连接")
except TimeoutError:
print("连接超时")
except Exception as e:
print(f"未知错误: {e}")
# ❌ 直接拼接用户输入到字符串
request = f"GET /{user_input} HTTP/1.1\r\n" # 可能被注入
# ✅ 使用标准库编码 URL
from urllib.parse import quote
safe_path = quote(user_input)
request = f"GET /{safe_path} HTTP/1.1\r\n"
总结
本文系统讲解了 Python 网络编程的核心知识：
- TCP/IP 基础 — 分层模型、TCP vs UDP 对比
- Socket API — 创建/绑定/监听/accept/send/recv
- TCP 服务器 — 多线程处理、超时控制、ECHO 服务
- UDP 编程 — 无连接通信、广播、快速数据传输
- HTTP 协议 — 原始 Socket 发送 HTTP 请求、手动实现 HTTP 服务器
- 高级特性 — 非阻塞 I/O (select)、SSL/TLS 加密
- 实用工具 — 端口扫描器、聊天室、Banner 识别
本文是 Python 教程系列的第四十七篇。