Python网络编程
网络编程基础概念
1. 网络模型
OSI七层模型
- 物理层:传输比特流,负责物理连接和信号传输
- 数据链路层:传输帧,负责点到点的通信
- 网络层:传输数据包,负责路由选择和寻址
- 传输层:传输段或数据报,负责端到端的通信
- 会话层:负责会话管理和同步
- 表示层:负责数据格式转换和加密
- 应用层:提供应用程序接口,如HTTP、FTP、SMTP等
TCP/IP四层模型
- 网络接口层:对应OSI的物理层和数据链路层
- 网络层:对应OSI的网络层,如IP、ICMP、ARP等
- 传输层:对应OSI的传输层,如TCP、UDP等
- 应用层:对应OSI的会话层、表示层和应用层,如HTTP、FTP、SMTP等
2. IP地址和端口
- IP地址:标识网络中的设备,如192.168.1.1
- 端口:标识设备上的进程,范围0-65535
- 0-1023:知名端口(如HTTP 80,HTTPS 443,FTP 21等)
- 1024-49151:注册端口
- 49152-65535:动态端口
3. 传输层协议
TCP(传输控制协议)
- 面向连接的协议
- 可靠的、有序的、基于字节流的传输
- 三次握手建立连接,四次挥手关闭连接
- 拥塞控制和流量控制
- 适合对可靠性要求高的场景,如HTTP、FTP等
UDP(用户数据报协议)
- 无连接的协议
- 不可靠的、无序的、基于数据报的传输
- 没有拥塞控制和流量控制
- 适合对实时性要求高的场景,如视频流、语音通话等
4. 应用层协议
- HTTP(超文本传输协议):用于Web通信
- HTTPS(安全超文本传输协议):加密的HTTP
- FTP(文件传输协议):用于文件传输
- SMTP(简单邮件传输协议):用于发送邮件
- POP3(邮局协议版本3):用于接收邮件
- IMAP(互联网邮件访问协议):用于接收邮件,功能更强大
- DNS(域名系统):用于域名解析
- SSH(安全外壳协议):用于远程登录
Socket编程
1. Socket简介
Socket(套接字)是网络编程的基础,它提供了一套API,用于在不同设备上的进程之间进行通信。Socket可以使用TCP或UDP协议。
2. Socket的基本操作
python
import socket
# 创建Socket
s = socket.socket(family, type, proto=0)
# family: 地址族,如socket.AF_INET(IPv4)、socket.AF_INET6(IPv6)
# type: 套接字类型,如socket.SOCK_STREAM(TCP)、socket.SOCK_DGRAM(UDP)
# proto: 协议,默认为0
# 绑定地址和端口
s.bind((host, port))
# 监听连接(仅TCP)
s.listen(backlog)
# 接受连接(仅TCP)
conn, addr = s.accept()
# 连接服务器(仅TCP)
s.connect((host, port))
# 发送数据
s.send(bytes)
s.sendall(bytes) # 确保所有数据都被发送
# 接收数据
data = s.recv(bufsize)
# 关闭连接
s.close()import socket
# 创建Socket
s = socket.socket(family, type, proto=0)
# family: 地址族,如socket.AF_INET(IPv4)、socket.AF_INET6(IPv6)
# type: 套接字类型,如socket.SOCK_STREAM(TCP)、socket.SOCK_DGRAM(UDP)
# proto: 协议,默认为0
# 绑定地址和端口
s.bind((host, port))
# 监听连接(仅TCP)
s.listen(backlog)
# 接受连接(仅TCP)
conn, addr = s.accept()
# 连接服务器(仅TCP)
s.connect((host, port))
# 发送数据
s.send(bytes)
s.sendall(bytes) # 确保所有数据都被发送
# 接收数据
data = s.recv(bufsize)
# 关闭连接
s.close()TCP编程
1. TCP服务器
python
import socket
# 创建TCP服务器
def tcp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接,backlog=5表示最多允许5个连接等待
server_socket.listen(5)
print("TCP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
print(f"接收到来自 {client_addr} 的连接")
try:
# 接收数据,bufsize=1024表示每次最多接收1024字节
data = client_socket.recv(1024)
if data:
print(f"接收到数据: {data.decode('utf-8')}")
# 发送响应
client_socket.sendall(b"Hello, Client!")
finally:
# 关闭客户端连接
client_socket.close()
if __name__ == "__main__":
tcp_server()import socket
# 创建TCP服务器
def tcp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接,backlog=5表示最多允许5个连接等待
server_socket.listen(5)
print("TCP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
print(f"接收到来自 {client_addr} 的连接")
try:
# 接收数据,bufsize=1024表示每次最多接收1024字节
data = client_socket.recv(1024)
if data:
print(f"接收到数据: {data.decode('utf-8')}")
# 发送响应
client_socket.sendall(b"Hello, Client!")
finally:
# 关闭客户端连接
client_socket.close()
if __name__ == "__main__":
tcp_server()2. TCP客户端
python
import socket
# 创建TCP客户端
def tcp_client():
# 创建Socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接服务器
client_socket.connect(('127.0.0.1', 8080))
print("已连接到TCP服务器")
try:
# 发送数据
client_socket.sendall(b"Hello, Server!")
# 接收响应
data = client_socket.recv(1024)
print(f"接收到服务器响应: {data.decode('utf-8')}")
finally:
# 关闭连接
client_socket.close()
if __name__ == "__main__":
tcp_client()import socket
# 创建TCP客户端
def tcp_client():
# 创建Socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接服务器
client_socket.connect(('127.0.0.1', 8080))
print("已连接到TCP服务器")
try:
# 发送数据
client_socket.sendall(b"Hello, Server!")
# 接收响应
data = client_socket.recv(1024)
print(f"接收到服务器响应: {data.decode('utf-8')}")
finally:
# 关闭连接
client_socket.close()
if __name__ == "__main__":
tcp_client()3. 多线程TCP服务器
python
import socket
import threading
# 处理客户端连接的线程函数
def handle_client(client_socket, client_addr):
print(f"线程 {threading.current_thread().name} 处理来自 {client_addr} 的连接")
try:
while True:
# 接收数据
data = client_socket.recv(1024)
if not data:
break
print(f"接收到来自 {client_addr} 的数据: {data.decode('utf-8')}")
# 发送响应
client_socket.sendall(data)
finally:
# 关闭客户端连接
client_socket.close()
print(f"已关闭来自 {client_addr} 的连接")
# 创建多线程TCP服务器
def multi_threaded_tcp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接
server_socket.listen(5)
print("多线程TCP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
# 创建线程处理连接
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_addr),
daemon=True
)
client_thread.start()
if __name__ == "__main__":
multi_threaded_tcp_server()import socket
import threading
# 处理客户端连接的线程函数
def handle_client(client_socket, client_addr):
print(f"线程 {threading.current_thread().name} 处理来自 {client_addr} 的连接")
try:
while True:
# 接收数据
data = client_socket.recv(1024)
if not data:
break
print(f"接收到来自 {client_addr} 的数据: {data.decode('utf-8')}")
# 发送响应
client_socket.sendall(data)
finally:
# 关闭客户端连接
client_socket.close()
print(f"已关闭来自 {client_addr} 的连接")
# 创建多线程TCP服务器
def multi_threaded_tcp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接
server_socket.listen(5)
print("多线程TCP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
# 创建线程处理连接
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_addr),
daemon=True
)
client_thread.start()
if __name__ == "__main__":
multi_threaded_tcp_server()UDP编程
1. UDP服务器
python
import socket
# 创建UDP服务器
def udp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
print("UDP服务器已启动,监听端口8080...")
while True:
# 接收数据
data, client_addr = server_socket.recvfrom(1024)
print(f"接收到来自 {client_addr} 的数据: {data.decode('utf-8')}")
# 发送响应
server_socket.sendto(b"Hello, Client!", client_addr)
if __name__ == "__main__":
udp_server()import socket
# 创建UDP服务器
def udp_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
print("UDP服务器已启动,监听端口8080...")
while True:
# 接收数据
data, client_addr = server_socket.recvfrom(1024)
print(f"接收到来自 {client_addr} 的数据: {data.decode('utf-8')}")
# 发送响应
server_socket.sendto(b"Hello, Client!", client_addr)
if __name__ == "__main__":
udp_server()2. UDP客户端
python
import socket
# 创建UDP客户端
def udp_client():
# 创建Socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 发送数据
server_addr = ('127.0.0.1', 8080)
client_socket.sendto(b"Hello, Server!", server_addr)
# 接收响应
data, server_addr = client_socket.recvfrom(1024)
print(f"接收到服务器响应: {data.decode('utf-8')}")
# 关闭连接
client_socket.close()
if __name__ == "__main__":
udp_client()import socket
# 创建UDP客户端
def udp_client():
# 创建Socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# 发送数据
server_addr = ('127.0.0.1', 8080)
client_socket.sendto(b"Hello, Server!", server_addr)
# 接收响应
data, server_addr = client_socket.recvfrom(1024)
print(f"接收到服务器响应: {data.decode('utf-8')}")
# 关闭连接
client_socket.close()
if __name__ == "__main__":
udp_client()HTTP编程
1. 使用socket实现简单的HTTP服务器
python
import socket
# 简单的HTTP服务器
def http_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接
server_socket.listen(5)
print("HTTP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
try:
# 接收请求
request = client_socket.recv(1024).decode('utf-8')
print(f"接收到请求: {request}")
# 构建响应
response = """HTTP/1.1 200 OK\n"""
response += "Content-Type: text/html\n"
response += "\n"
response += "<h1>Hello, HTTP!</h1>"
# 发送响应
client_socket.sendall(response.encode('utf-8'))
finally:
# 关闭客户端连接
client_socket.close()
if __name__ == "__main__":
http_server()import socket
# 简单的HTTP服务器
def http_server():
# 创建Socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('127.0.0.1', 8080))
# 监听连接
server_socket.listen(5)
print("HTTP服务器已启动,监听端口8080...")
while True:
# 接受连接
client_socket, client_addr = server_socket.accept()
try:
# 接收请求
request = client_socket.recv(1024).decode('utf-8')
print(f"接收到请求: {request}")
# 构建响应
response = """HTTP/1.1 200 OK\n"""
response += "Content-Type: text/html\n"
response += "\n"
response += "<h1>Hello, HTTP!</h1>"
# 发送响应
client_socket.sendall(response.encode('utf-8'))
finally:
# 关闭客户端连接
client_socket.close()
if __name__ == "__main__":
http_server()2. 使用urllib库
python
from urllib import request, error
# 发送GET请求
try:
response = request.urlopen('https://www.baidu.com')
print(f"状态码: {response.status}")
print(f"响应头: {response.getheaders()}")
content = response.read().decode('utf-8')
print(f"响应内容长度: {len(content)}")
except error.HTTPError as e:
print(f"HTTP错误: {e.code} {e.reason}")
except error.URLError as e:
print(f"URL错误: {e.reason}")
# 发送POST请求
import urllib.parse
data = {
'name': '张三',
'age': 20
}
data = urllib.parse.urlencode(data).encode('utf-8')
req = request.Request('https://httpbin.org/post', data=data)
req.add_header('User-Agent', 'Mozilla/5.0')
with request.urlopen(req) as response:
print(response.read().decode('utf-8'))from urllib import request, error
# 发送GET请求
try:
response = request.urlopen('https://www.baidu.com')
print(f"状态码: {response.status}")
print(f"响应头: {response.getheaders()}")
content = response.read().decode('utf-8')
print(f"响应内容长度: {len(content)}")
except error.HTTPError as e:
print(f"HTTP错误: {e.code} {e.reason}")
except error.URLError as e:
print(f"URL错误: {e.reason}")
# 发送POST请求
import urllib.parse
data = {
'name': '张三',
'age': 20
}
data = urllib.parse.urlencode(data).encode('utf-8')
req = request.Request('https://httpbin.org/post', data=data)
req.add_header('User-Agent', 'Mozilla/5.0')
with request.urlopen(req) as response:
print(response.read().decode('utf-8'))3. 使用requests库
bash
pip install requestspip install requestspython
import requests
# 发送GET请求
response = requests.get('https://www.baidu.com')
print(f"状态码: {response.status_code}")
print(f"响应头: {response.headers}")
print(f"响应内容: {response.text[:100]}...")
# 发送POST请求
data = {
'name': '张三',
'age': 20
}
response = requests.post('https://httpbin.org/post', data=data)
print(f"POST响应: {response.json()}")
# 添加请求头
headers = {
'User-Agent': 'Mozilla/5.0',
'Authorization': 'Bearer token'
}
response = requests.get('https://api.example.com', headers=headers)
# 添加查询参数
params = {
'page': 1,
'limit': 10
}
response = requests.get('https://api.example.com/users', params=params)
# 文件上传
files = {
'file': open('file.txt', 'rb')
}
response = requests.post('https://httpbin.org/post', files=files)
# 会话管理
session = requests.Session()
session.get('https://httpbin.org/cookies/set/sessioncookie/123456')
response = session.get('https://httpbin.org/cookies')
print(f"Cookie: {response.json()}")import requests
# 发送GET请求
response = requests.get('https://www.baidu.com')
print(f"状态码: {response.status_code}")
print(f"响应头: {response.headers}")
print(f"响应内容: {response.text[:100]}...")
# 发送POST请求
data = {
'name': '张三',
'age': 20
}
response = requests.post('https://httpbin.org/post', data=data)
print(f"POST响应: {response.json()}")
# 添加请求头
headers = {
'User-Agent': 'Mozilla/5.0',
'Authorization': 'Bearer token'
}
response = requests.get('https://api.example.com', headers=headers)
# 添加查询参数
params = {
'page': 1,
'limit': 10
}
response = requests.get('https://api.example.com/users', params=params)
# 文件上传
files = {
'file': open('file.txt', 'rb')
}
response = requests.post('https://httpbin.org/post', files=files)
# 会话管理
session = requests.Session()
session.get('https://httpbin.org/cookies/set/sessioncookie/123456')
response = session.get('https://httpbin.org/cookies')
print(f"Cookie: {response.json()}")异步网络编程
1. 使用asyncio和aiohttp
bash
pip install aiohttppip install aiohttppython
import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
status = response.status
content = await response.text()
print(f"URL: {url}, Status: {status}, Content Length: {len(content)}")
return status
async def main():
urls = [
"https://www.baidu.com",
"https://www.google.com",
"https://www.github.com"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"所有请求完成,结果: {results}")
asyncio.run(main())import asyncio
import aiohttp
async def fetch_url(session, url):
async with session.get(url) as response:
status = response.status
content = await response.text()
print(f"URL: {url}, Status: {status}, Content Length: {len(content)}")
return status
async def main():
urls = [
"https://www.baidu.com",
"https://www.google.com",
"https://www.github.com"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
results = await asyncio.gather(*tasks)
print(f"所有请求完成,结果: {results}")
asyncio.run(main())2. 使用Tornado框架
bash
pip install tornadopip install tornadopython
import tornado.ioloop
import tornado.web
import tornado.websocket
# 定义请求处理类
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, Tornado!")
class UserHandler(tornado.web.RequestHandler):
def get(self, user_id):
self.write(f"User ID: {user_id}")
# WebSocket处理类
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def open(self):
print("WebSocket连接已打开")
def on_message(self, message):
print(f"接收到消息: {message}")
self.write_message(f"服务器回复: {message}")
def on_close(self):
print("WebSocket连接已关闭")
# 定义路由
urls = [
(r"/", MainHandler),
(r"/user/(\d+)", UserHandler),
(r"/ws", WebSocketHandler),
]
# 创建应用
app = tornado.web.Application(urls)
if __name__ == "__main__":
# 监听端口
app.listen(8080)
print("Tornado服务器已启动,监听端口8080...")
# 启动事件循环
tornado.ioloop.IOLoop.current().start()import tornado.ioloop
import tornado.web
import tornado.websocket
# 定义请求处理类
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, Tornado!")
class UserHandler(tornado.web.RequestHandler):
def get(self, user_id):
self.write(f"User ID: {user_id}")
# WebSocket处理类
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def open(self):
print("WebSocket连接已打开")
def on_message(self, message):
print(f"接收到消息: {message}")
self.write_message(f"服务器回复: {message}")
def on_close(self):
print("WebSocket连接已关闭")
# 定义路由
urls = [
(r"/", MainHandler),
(r"/user/(\d+)", UserHandler),
(r"/ws", WebSocketHandler),
]
# 创建应用
app = tornado.web.Application(urls)
if __name__ == "__main__":
# 监听端口
app.listen(8080)
print("Tornado服务器已启动,监听端口8080...")
# 启动事件循环
tornado.ioloop.IOLoop.current().start()使用 httpx(现代 HTTP 客户端)
httpx 是 requests 的现代替代品,同时支持同步和异步:
bash
pip install httpxpip install httpxpython
import httpx
# 同步请求
with httpx.Client() as client:
response = client.get("https://api.github.com")
print(response.status_code)
print(response.json())
# POST
response = client.post("https://httpbin.org/post", json={"key": "value"})
# 超时设置
response = client.get("https://example.com", timeout=30.0)
# 异步请求
async def async_fetch():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.github.com")
return response.json()
# 连接池
with httpx.Client() as client:
# 自动管理连接池,复用连接
for i in range(10):
r = client.get(f"https://api.example.com/page/{i}")
# 对比 requests vs httpx
# - httpx 支持 HTTP/2(需安装 h2)
# - httpx 原生支持异步
# - httpx 有更现代的 API 设计
# - requests 生态更成熟,兼容性更好import httpx
# 同步请求
with httpx.Client() as client:
response = client.get("https://api.github.com")
print(response.status_code)
print(response.json())
# POST
response = client.post("https://httpbin.org/post", json={"key": "value"})
# 超时设置
response = client.get("https://example.com", timeout=30.0)
# 异步请求
async def async_fetch():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.github.com")
return response.json()
# 连接池
with httpx.Client() as client:
# 自动管理连接池,复用连接
for i in range(10):
r = client.get(f"https://api.example.com/page/{i}")
# 对比 requests vs httpx
# - httpx 支持 HTTP/2(需安装 h2)
# - httpx 原生支持异步
# - httpx 有更现代的 API 设计
# - requests 生态更成熟,兼容性更好网络编程的最佳实践
- 错误处理:捕获和处理网络异常,如连接超时、连接拒绝等
- 资源管理:确保及时关闭Socket连接,释放资源
- 线程安全:多线程环境下,确保Socket的使用是线程安全的
- 超时设置:设置合理的超时时间,避免无限期等待
- 编码处理:正确处理字符串编码,避免乱码
- 日志记录:记录网络请求和响应,便于调试和监控
- 安全性:使用加密协议(如HTTPS),避免明文传输敏感数据
- 性能优化:使用连接池、批量操作等优化网络性能
- 并发处理:根据任务类型选择合适的并发模型(多线程、多进程、异步编程)
- 遵循协议规范:严格遵循网络协议的规范,确保兼容性
常见网络编程库
- socket:Python标准库,提供底层的网络编程API
- urllib:Python标准库,提供HTTP客户端功能
- requests:第三方库,更简洁易用的HTTP客户端
- aiohttp:第三方库,异步HTTP客户端和服务器
- tornado:第三方库,异步Web框架和网络库
- twisted:第三方库,事件驱动的网络编程框架
- paramiko:第三方库,SSH协议实现
- scapy:第三方库,网络数据包处理库
- pycurl:第三方库,基于libcurl的HTTP客户端
- websocket-client:第三方库,WebSocket客户端
总结
Python提供了丰富的网络编程库,从底层的socket库到高层的Web框架,满足不同场景的需求。在实际开发中,我们需要根据具体的需求选择合适的库和协议:
- 对于简单的网络通信,使用socket库
- 对于HTTP通信,使用requests库或aiohttp库
- 对于Web应用开发,使用Django、Flask或FastAPI框架
- 对于实时通信,使用WebSocket
- 对于高性能的网络应用,使用异步编程
网络编程是Python开发中的重要组成部分,掌握网络编程的基本概念和常用库,可以帮助我们开发各种网络应用,如Web服务器、API服务、爬虫、实时通信应用等。
Cao Note