Python 网络编程 socket/http
Python 网络编程 socket/http 的完整教学讲解。
socket TCP 服务端
基本写法:创建 TCP 服务端
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 创建 IPv4 TCP 套接字并监听
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1", 8080))
s.listen(5)
conn, addr = s.accept()
data = conn.recv(1024)
conn.sendall(b"hello")
conn.close()
基本写法:设置地址复用
<套接字>.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 避免端口释放等待,立即重启绑定
s = socket.socket()
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("", 8080))
基本写法:并发接收连接
while True: <套接字>.accept()
# 循环接受多个客户端连接
while True:
conn, addr = s.accept()
try:
while True:
data = conn.recv(1024)
if not data:
break
conn.sendall(data) # 回显
finally:
conn.close()
基本写法:设置超时
<套接字>.settimeout(<秒数>)
# 设置阻塞操作的超时时间
s.settimeout(5.0)
try:
conn, addr = s.accept()
except socket.timeout:
print("接受连接超时")
socket TCP 客户端
基本写法:创建 TCP 客户端
<套接字>.connect((<主机>, <端口>))
# 连接服务端并发送数据
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("127.0.0.1", 8080))
s.sendall(b"ping")
response = s.recv(1024)
s.close()
基本写法:发送字符串数据
<字符串>.encode(<编码>)
# 字符串转字节后发送
s.sendall("你好".encode("utf-8"))
response = s.recv(1024).decode("utf-8")
socket UDP 通信
基本写法:UDP 服务端
socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# UDP 无连接,直接接收数据报
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.bind(("127.0.0.1", 8080))
data, addr = s.recvfrom(1024)
s.sendto(b"ack", addr)
基本写法:UDP 客户端
<套接字>.sendto(<数据>, (<主机>, <端口>))
# UDP 发送无需建立连接
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.sendto(b"hello", ("127.0.0.1", 8080))
data, addr = s.recvfrom(1024)
socketserver 模块
基本写法:TCP 请求处理类
class <类>(socketserver.BaseRequestHandler):
# 继承 BaseRequestHandler 简化服务端
import socketserver
class Handler(socketserver.BaseRequestHandler):
def handle(self):
data = self.request.recv(1024)
self.request.sendall(data)
server = socketserver.TCPServer(("127.0.0.1", 8080), Handler)
server.serve_forever()
基本写法:多线程服务端
socketserver.ThreadingTCPServer(<地址>, <处理器>)
# 每个连接独立线程处理
server = socketserver.ThreadingTCPServer(("0.0.0.0", 8080), Handler)
server.serve_forever()
基本写法:UDP 服务端
socketserver.UDPServer(<地址>, <处理器>)
# UDP 请求处理
class UDPHandler(socketserver.BaseRequestHandler):
def handle(self):
data, sock = self.request
sock.sendto(b"ack", self.client_address)
server = socketserver.UDPServer(("127.0.0.1", 8080), UDPHandler)
http.client 标准库客户端
基本写法:HTTP GET 请求
http.client.HTTPSConnection(<主机>)
# 发起 HTTPS GET 请求
import http.client
conn = http.client.HTTPSConnection("www.example.com")
conn.request("GET", "/")
resp = conn.getresponse()
print(resp.status, resp.read().decode())
conn.close()
基本写法:带请求头
conn.request(<方法>, <路径>, <请求体>, <头字典>)
# 携带自定义请求头
headers = {"Authorization": "Bearer token123"}
conn.request("GET", "/api", headers=headers)
基本写法:HTTP POST 请求
conn.request("POST", <路径>, <请求体>)
# 发送 POST 请求体
import json
body = json.dumps({"name": "Tom"})
conn.request("POST", "/api", body, {"Content-Type": "application/json"})
urllib.request 请求
基本写法:发起 GET 请求
urllib.request.urlopen(<URL>)
# 最简方式打开 URL
from urllib.request import urlopen
resp = urlopen("https://www.example.com")
html = resp.read().decode("utf-8")
基本写法:构造 Request 对象
urllib.request.Request(<URL>, headers=<头>)
# 自定义请求头
from urllib.request import Request, urlopen
req = Request("https://example.com", headers={"User-Agent": "MyApp"})
resp = urlopen(req)
基本写法:POST 请求
urlopen(<Request>, data=<字节串>)
# 发送表单数据
from urllib.parse import urlencode
data = urlencode({"q": "python"}).encode()
resp = urlopen(Request("https://example.com", data=data))
urllib.parse URL 处理
基本写法:URL 编码
urllib.parse.urlencode(<字典>)
# 字典转查询字符串
from urllib.parse import urlencode
print(urlencode({"a": 1, "b": "中文"})) # a=1&b=%E4%B8%AD%E6%96%87
基本写法:解析 URL
urllib.parse.urlparse(<URL>)
# 拆解 URL 各部分
from urllib.parse import urlparse
r = urlparse("https://a.com/path?q=1#frag")
print(r.scheme, r.netloc, r.path) # https a.com /path
基本写法:拼接 URL
urllib.parse.urljoin(<基础URL>, <相对路径>)
# 基于基础 URL 拼接相对路径
from urllib.parse import urljoin
print(urljoin("https://a.com/dir/", "page.html")) # https://a.com/dir/page.html
基本写法:解析查询字符串
urllib.parse.parse_qs(<查询串>)
# 查询字符串转字典
from urllib.parse import parse_qs
print(parse_qs("a=1&b=2&b=3")) # {'a': ['1'], 'b': ['2', '3']}
urllib.error 异常处理
基本写法:捕获 HTTP 错误
urllib.error.HTTPError
# 处理 HTTP 状态码错误
from urllib.request import urlopen
from urllib.error import HTTPError, URLError
try:
resp = urlopen("https://example.com/404")
except HTTPError as e:
print(e.code, e.reason) # 404 Not Found
except URLError as e:
print(e.reason)