网站首页 > 教程文章 正文
今日目标
o 理解网络编程的基本概念和协议
o 掌握Socket编程的基础知识
o 学会创建HTTP客户端和服务器
o 了解网络编程的安全考虑
网络编程概述
网络编程是计算机之间进行通信的基础，Python提供了丰富的网络编程库：
o socket：底层网络通信
o http.client：HTTP客户端
o urllib：URL处理
o requests：高级HTTP库
o asyncio：异步网络编程
网络协议层次
应用层 HTTP, FTP, SMTP
传输层 TCP, UDP
网络层 IP
链路层 以太网, WiFi
物理层 电缆, 光纤Socket编程基础
1. 创建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类型说明
# AF_INET: IPv4地址族
# AF_INET6: IPv6地址族
# SOCK_STREAM: TCP协议
# SOCK_DGRAM: UDP协议2. TCP客户端示例
import socket
import time
def tcp_client():
"""TCP客户端示例"""
# 创建Socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
# 连接服务器
server_address = ('localhost', 8888)
print(f"连接到服务器: {server_address}")
client.connect(server_address)
# 发送数据
message = "Hello, Server!"
client.send(message.encode('utf-8'))
print(f"发送: {message}")
# 接收响应
data = client.recv(1024)
response = data.decode('utf-8')
print(f"收到: {response}")
except ConnectionRefusedError:
print("连接被拒绝，请确保服务器正在运行")
except Exception as e:
print(f"连接错误: {e}")
finally:
client.close()
print("连接已关闭")
# 运行客户端
if __name__ == "__main__":
tcp_client()3. TCP服务器示例
import socket
import threading
import time
class TCPServer:
def __init__(self, host='localhost', port=8888):
self.host = host
self.port = port
self.server_socket = None
self.clients = []
def start(self):
"""启动服务器"""
try:
# 创建服务器Socket
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置端口重用
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 绑定地址
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()
print(f"新客户端连接: {client_address}")
# 为每个客户端创建新线程
client_thread = threading.Thread(
target=self.handle_client,
args=(client_socket, client_address)
)
client_thread.start()
self.clients.append(client_socket)
except Exception as e:
print(f"服务器错误: {e}")
finally:
self.stop()
def handle_client(self, client_socket, client_address):
"""处理客户端连接"""
try:
while True:
# 接收数据
data = client_socket.recv(1024)
if not data:
break
# 解码数据
message = data.decode('utf-8')
print(f"来自 {client_address}: {message}")
# 发送响应
response = f"服务器收到: {message}"
client_socket.send(response.encode('utf-8'))
except Exception as e:
print(f"处理客户端 {client_address} 时出错: {e}")
finally:
client_socket.close()
if client_socket in self.clients:
self.clients.remove(client_socket)
print(f"客户端 {client_address} 断开连接")
def stop(self):
"""停止服务器"""
if self.server_socket:
self.server_socket.close()
for client in self.clients:
client.close()
print("服务器已停止")
# 启动服务器
if __name__ == "__main__":
server = TCPServer()
try:
server.start()
except KeyboardInterrupt:
print("\n正在关闭服务器...")
server.stop()HTTP编程
1. 使用urllib
import urllib.request
import urllib.parse
import urllib.error
import json
def http_get_example():
"""HTTP GET请求示例"""
try:
# 发送GET请求
url = "https://jsonplaceholder.typicode.com/posts/1"
with urllib.request.urlopen(url) as response:
data = response.read()
result = json.loads(data.decode('utf-8'))
print("GET请求结果:")
print(f"标题: {result['title']}")
print(f"内容: {result['body']}")
except urllib.error.URLError as e:
print(f"URL错误: {e}")
except Exception as e:
print(f"请求错误: {e}")
def http_post_example():
"""HTTP POST请求示例"""
try:
# 准备数据
post_data = {
"title": "Python网络编程",
"body": "学习Python网络编程很有趣",
"userId": 1
}
# 编码数据
data = json.dumps(post_data).encode('utf-8')
# 创建请求
url = "https://jsonplaceholder.typicode.com/posts"
req = urllib.request.Request(
url,
data=data,
headers={'Content-Type': 'application/json'}
)
# 发送请求
with urllib.request.urlopen(req) as response:
result = json.loads(response.read().decode('utf-8'))
print("POST请求结果:")
print(f"新帖子ID: {result['id']}")
print(f"标题: {result['title']}")
except urllib.error.URLError as e:
print(f"URL错误: {e}")
except Exception as e:
print(f"请求错误: {e}")
# 运行示例
http_get_example()
print("-" * 50)
http_post_example()2. 使用requests库
import requests
import json
class HTTPClient:
def __init__(self, base_url="https://jsonplaceholder.typicode.com"):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Python-HTTPClient/1.0',
'Content-Type': 'application/json'
})
def get(self, endpoint, params=None):
"""发送GET请求"""
try:
url = f"{self.base_url}{endpoint}"
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"GET请求错误: {e}")
return None
def post(self, endpoint, data=None):
"""发送POST请求"""
try:
url = f"{self.base_url}{endpoint}"
response = self.session.post(url, json=data, timeout=10)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"POST请求错误: {e}")
return None
def put(self, endpoint, data=None):
"""发送PUT请求"""
try:
url = f"{self.base_url}{endpoint}"
response = self.session.put(url, json=data, timeout=10)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
print(f"PUT请求错误: {e}")
return None
def delete(self, endpoint):
"""发送DELETE请求"""
try:
url = f"{self.base_url}{endpoint}"
response = self.session.delete(url, timeout=10)
response.raise_for_status()
return response.status_code == 204
except requests.RequestException as e:
print(f"DELETE请求错误: {e}")
return False
# 使用示例
client = HTTPClient()
# 获取所有帖子
posts = client.get("/posts")
if posts:
print(f"获取到 {len(posts)} 个帖子")
for post in posts[:3]:
print(f"- {post['title']}")
# 创建新帖子
new_post = {
"title": "Python网络编程教程",
"body": "这是一个关于Python网络编程的详细教程",
"userId": 1
}
created_post = client.post("/posts", new_post)
if created_post:
print(f"新帖子创建成功，ID: {created_post['id']}")
# 更新帖子
update_data = {"title": "更新的标题"}
updated_post = client.put(f"/posts/{created_post['id']}", update_data)
if updated_post:
print(f"帖子更新成功: {updated_post['title']}")
# 删除帖子
if client.delete(f"/posts/{created_post['id']}"):
print("帖子删除成功")网络编程安全
1. 输入验证
import re
import socket
def validate_ip_address(ip):
"""验证IP地址格式"""
pattern = r'^(\d{1,3}\.){3}\d{1,3}#39;
if not re.match(pattern, ip):
return False
parts = ip.split('.')
for part in parts:
if not 0 <= int(part) <= 255:
return False
return True
def validate_port(port):
"""验证端口号"""
try:
port_num = int(port)
return 1 <= port_num <= 65535
except ValueError:
return False
def safe_socket_connect(host, port):
"""安全的Socket连接"""
if not validate_ip_address(host):
raise ValueError(f"无效的IP地址: {host}")
if not validate_port(port):
raise ValueError(f"无效的端口号: {port}")
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10) # 设置超时
sock.connect((host, port))
return sock
except socket.timeout:
raise ConnectionError("连接超时")
except socket.error as e:
raise ConnectionError(f"连接失败: {e}")2. 数据加密
import ssl
import socket
def create_ssl_client(host, port):
"""创建SSL客户端"""
try:
# 创建SSL上下文
context = ssl.create_default_context()
# 创建Socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 包装为SSL Socket
ssl_sock = context.wrap_socket(sock, server_hostname=host)
# 连接
ssl_sock.connect((host, port))
return ssl_sock
except ssl.SSLError as e:
print(f"SSL错误: {e}")
return None
except Exception as e:
print(f"连接错误: {e}")
return None
def create_ssl_server(host, port, cert_file, key_file):
"""创建SSL服务器"""
try:
# 创建SSL上下文
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(cert_file, key_file)
# 创建Socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind((host, port))
sock.listen(5)
# 包装为SSL Socket
ssl_sock = context.wrap_socket(sock, server_side=True)
return ssl_sock
except Exception as e:
print(f"SSL服务器创建失败: {e}")
return None今日总结
今天我们学习了网络编程的基础知识：
1. Socket编程：TCP和UDP通信的基础
2. HTTP编程：使用urllib和requests进行HTTP请求
3. 网络安全：输入验证、SSL加密等安全考虑
4. 实用示例：客户端和服务器开发
网络编程是Python应用开发的重要组成部分，掌握这些知识对于开发Web应用、网络工具、分布式系统等都至关重要。
猜你喜欢
- 2025-08-05 浅谈ActiveMQ与使用
- 2025-08-05 Chinese premier calls for forging example of openness, development cooperation with ASEAN, GCC
- 2025-08-05 西门子博途有关通过 PUT/GET 指令通信的基本信息
- 2025-08-05 数据源连接池的原理及 Tomcat 中的应用
- 2025-08-05 ActiveMQ实现站内消息提醒功能
- 2025-08-05 Android 传统(经典)蓝牙框架
- 2025-08-05 前端测试新范式:Vitest+Playwright 如何让测试效率提升 400%
- 2025-08-05 MCP 客户端连接与请求流程深度解析
- 2025-08-05 亚马逊云代理商:怎样使用Glue构建ETL管道?
- 2025-08-05 一个基础又很重要的知识点:JDBC原理(基本案例和面试知识点)
- 08-05 Docker Compose 编排实战:一键部署多容器应用!
- 08-05 Docker 命令入门实战:搞懂这些才算真正入门!
- 08-05Docker 镜像构建加速与镜像瘦身实战，一篇就够!
- 08-05Docker 常用命令手册
- 08-05Docker命令大全详解(39个常用命令)
- 08-05Docker镜像越来越大?我用这 3 个技巧直接瘦身 80%!附实战优化对比
- 08-05Docker容器与镜像详解(6大核心区别)
- 08-05docker镜像操作
- 最近发表
- 标签列表
-
- location.href (44)
- document.ready (36)
- git checkout -b (34)
- 跃点数 (35)
- 阿里云镜像地址 (33)
- qt qmessagebox (36)
- mybatis plus page (35)
- vue @scroll (38)
- 堆栈区别 (33)
- 什么是容器 (33)
- sha1 md5 (33)
- navicat导出数据 (34)
- 阿里云acp考试 (33)
- 阿里云 nacos (34)
- redhat官网下载镜像 (36)
- srs服务器 (33)
- pico开发者 (33)
- https的端口号 (34)
- vscode更改主题 (35)
- 阿里云资源池 (34)
- os.path.join (33)
- redis aof rdb 区别 (33)
- 302跳转 (33)
- http method (35)
- js array splice (33)