当前位置： 首页 > news >正文

Python 第二十四节 Pythong中网络编程相关详细使用及案例

news 2025/11/1 12:04:21

前言

主要分为以下几个方面进行介绍说明： 网络编程基础 、Socket 编程 、TCP 编程、UDP 编程、HTTP 编程、高级网络编程；

1、注意事项

网络编程基础
网络协议层次
应用层: HTTP, FTP, SMTP
传输层: TCP, UDP
网络层: IP, ICMP
数据链路层: Ethernet

2、核心概念

  1. IP地址: 网络设备的唯一标识
  2. 端口: 应用程序的通信端点
  3. Socket: 网络通信的端点

2.1、Socket 编程

创建 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 方法

# 服务器端方法
socket.bind((host, port))    # 绑定地址
socket.listen(backlog)       # 开始监听
socket.accept()             # 接受连接# 客户端方法
socket.connect((host, port)) # 连接服务器# 通用方法
socket.send(data)           # 发送数据
socket.recv(bufsize)        # 接收数据
socket.close()              # 关闭连接

2.2、TCP 编程

TCP 服务器

import socket
import threadingdef handle_client(client_socket, address):"""处理客户端连接"""print(f"连接来自: {address}")try:while True:# 接收数据data = client_socket.recv(1024)if not data:breakmessage = data.decode('utf-8')print(f"收到来自 {address} 的消息: {message}")# 发送响应response = f"服务器已收到: {message}"client_socket.send(response.encode('utf-8'))except Exception as e:print(f"处理客户端 {address} 时出错: {e}")finally:client_socket.close()print(f"与 {address} 的连接已关闭")def tcp_server(host='localhost', port=8888):"""TCP服务器实现"""server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 设置端口复用server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)try:server_socket.bind((host, port))server_socket.listen(5)print(f"TCP服务器启动在 {host}:{port}")while True:client_socket, address = server_socket.accept()# 为每个客户端创建新线程client_thread = threading.Thread(target=handle_client, args=(client_socket, address))client_thread.daemon = Trueclient_thread.start()except KeyboardInterrupt:print("\n服务器关闭")except Exception as e:print(f"服务器错误: {e}")finally:server_socket.close()if __name__ == "__main__":tcp_server()

TCP 客户端

import socketdef tcp_client(host='localhost', port=8888):"""TCP客户端实现"""client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)try:client_socket.connect((host, port))print(f"已连接到服务器 {host}:{port}")while True:# 获取用户输入message = input("请输入消息 (输入 'quit' 退出): ")if message.lower() == 'quit':break# 发送消息client_socket.send(message.encode('utf-8'))# 接收响应response = client_socket.recv(1024)print(f"服务器响应: {response.decode('utf-8')}")except Exception as e:print(f"客户端错误: {e}")finally:client_socket.close()print("连接已关闭")if __name__ == "__main__":tcp_client()

2.3、UDP 编程

UDP 服务器

import socketdef udp_server(host='localhost', port=8888):"""UDP服务器实现"""server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)try:server_socket.bind((host, port))print(f"UDP服务器启动在 {host}:{port}")while True:# 接收数据和客户端地址data, client_address = server_socket.recvfrom(1024)message = data.decode('utf-8')print(f"收到来自 {client_address} 的消息: {message}")# 发送响应response = f"UDP服务器已收到: {message}"server_socket.sendto(response.encode('utf-8'), client_address)except KeyboardInterrupt:print("\n服务器关闭")except Exception as e:print(f"服务器错误: {e}")finally:server_socket.close()if __name__ == "__main__":udp_server()

UDP 客户端

import socketdef udp_client(host='localhost', port=8888):"""UDP客户端实现"""client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)try:while True:# 获取用户输入message = input("请输入消息 (输入 'quit' 退出): ")if message.lower() == 'quit':break# 发送消息client_socket.sendto(message.encode('utf-8'), (host, port))# 接收响应response, server_address = client_socket.recvfrom(1024)print(f"服务器响应: {response.decode('utf-8')}")except Exception as e:print(f"客户端错误: {e}")finally:client_socket.close()if __name__ == "__main__":udp_client()

3、 HTTP 编程

使用 requests 库

import requests
import jsondef http_examples():"""HTTP请求示例"""# GET 请求print("=== GET 请求示例 ===")response = requests.get('https://httpbin.org/get')print(f"状态码: {response.status_code}")print(f"响应内容: {response.text}")# POST 请求print("\n=== POST 请求示例 ===")data = {'key': 'value', 'name': 'Python'}response = requests.post('https://httpbin.org/post', data=data)print(f"状态码: {response.status_code}")print(f"响应内容: {response.text}")# 带JSON的POST请求print("\n=== JSON POST 请求示例 ===")json_data = {'name': 'Alice', 'age': 25}response = requests.post('https://httpbin.org/post', json=json_data,headers={'Content-Type': 'application/json'})print(f"状态码: {response.status_code}")print(f"响应内容: {response.text}")# 带认证的请求print("\n=== 带认证的请求示例 ===")response = requests.get('https://httpbin.org/basic-auth/user/passwd',auth=('user', 'passwd'))print(f"状态码: {response.status_code}")print(f"响应内容: {response.text}")# 处理超时print("\n=== 超时处理示例 ===")try:response = requests.get('https://httpbin.org/delay/5', timeout=3)except requests.exceptions.Timeout:print("请求超时！")# 下载文件print("\n=== 文件下载示例 ===")response = requests.get('https://httpbin.org/image/jpeg')with open('image.jpg', 'wb') as f:f.write(response.content)print("文件下载完成")if __name__ == "__main__":http_examples()

简单的HTTP服务器

from http.server import HTTPServer, BaseHTTPRequestHandler
import jsonclass SimpleHTTPRequestHandler(BaseHTTPRequestHandler):def do_GET(self):"""处理GET请求"""if self.path == '/':self.send_response(200)self.send_header('Content-type', 'text/html')self.end_headers()self.wfile.write(b'<h1>欢迎来到Python HTTP服务器</h1>')elif self.path == '/api/data':self.send_response(200)self.send_header('Content-type', 'application/json')self.end_headers()data = {'message': 'Hello World', 'status': 'success'}self.wfile.write(json.dumps(data).encode())else:self.send_response(404)self.end_headers()self.wfile.write(b'404 - 页面未找到')def do_POST(self):"""处理POST请求"""if self.path == '/api/echo':content_length = int(self.headers['Content-Length'])post_data = self.rfile.read(content_length)self.send_response(200)self.send_header('Content-type', 'application/json')self.end_headers()response = {'received': post_data.decode('utf-8'),'length': content_length}self.wfile.write(json.dumps(response).encode())else:self.send_response(404)self.end_headers()def run_http_server(host='localhost', port=8000):"""运行HTTP服务器"""server_address = (host, port)httpd = HTTPServer(server_address, SimpleHTTPRequestHandler)print(f"HTTP服务器运行在 http://{host}:{port}")try:httpd.serve_forever()except KeyboardInterrupt:print("\n服务器关闭")httpd.shutdown()if __name__ == "__main__":run_http_server()

4、高级网络编程

4.1、异步网络编程 with asyncio

import asyncio
import aiohttpasync def fetch_url(session, url):"""异步获取URL内容"""try:async with session.get(url) as response:content = await response.text()return f"{url}: {len(content)} 字符"except Exception as e:return f"{url}: 错误 - {e}"async def async_http_requests():"""异步HTTP请求示例"""urls = ['https://httpbin.org/get','https://httpbin.org/delay/1','https://httpbin.org/status/200','https://httpbin.org/status/404']async with aiohttp.ClientSession() as session:tasks = [fetch_url(session, url) for url in urls]results = await asyncio.gather(*tasks)for result in results:print(result)async def async_tcp_echo_client(message, host='localhost', port=8888):"""异步TCP客户端"""reader, writer = await asyncio.open_connection(host, port)print(f'发送: {message}')writer.write(message.encode())await writer.drain()data = await reader.read(100)print(f'收到: {data.decode()}')writer.close()await writer.wait_closed()async def main():"""主异步函数"""print("=== 异步HTTP请求 ===")await async_http_requests()print("\n=== 异步TCP客户端 ===")await async_tcp_echo_client('Hello Asyncio!')if __name__ == "__main__":asyncio.run(main())

4.2、WebSocket 编程

import asyncio
import websocketsasync def websocket_echo(websocket, path):"""WebSocket回声服务器"""print("客户端连接")try:async for message in websocket:print(f"收到消息: {message}")await websocket.send(f"服务器回声: {message}")except websockets.exceptions.ConnectionClosed:print("客户端断开连接")async def websocket_client():"""WebSocket客户端"""uri = "ws://localhost:8765"async with websockets.connect(uri) as websocket:# 发送消息await websocket.send("Hello WebSocket!")print("已发送: Hello WebSocket!")# 接收响应response = await websocket.recv()print(f"收到: {response}")async def run_websocket_example():"""运行WebSocket示例"""# 启动服务器server = await websockets.serve(websocket_echo, "localhost", 8888)print("WebSocket服务器运行在 ws://localhost:/8888")# 等待一会儿后启动客户端await asyncio.sleep(1)await websocket_client()# 保持服务器运行await asyncio.Future()  # 永远运行if __name__ == "__main__":asyncio.run(run_websocket_example())

5、Python网络编程注意事项

5.1、 错误处理

import socket
import errnodef robust_socket_example():"""健壮的Socket编程示例"""try:# 创建socketsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 设置超时sock.settimeout(10.0)# 尝试连接sock.connect(('example.com', 80))# 发送数据sock.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')# 接收数据data = sock.recv(4096)print(f"收到数据: {data.decode()}")except socket.timeout:print("连接超时")except socket.gaierror as e:print(f"地址解析错误: {e}")except socket.error as e:if e.errno == errno.ECONNREFUSED:print("连接被拒绝")else:print(f"Socket错误: {e}")except Exception as e:print(f"未知错误: {e}")finally:# 确保关闭socketif 'sock' in locals():sock.close()

5.2、 安全性考虑

import ssldef secure_socket_example():"""安全Socket连接示例"""try:# 创建普通socketsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 包装为SSL socketcontext = ssl.create_default_context()# 验证证书context.check_hostname = Truecontext.verify_mode = ssl.CERT_REQUIREDsecure_sock = context.wrap_socket(sock, server_hostname='example.com')secure_sock.connect(('example.com', 443))# 发送HTTPS请求secure_sock.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')response = secure_sock.recv(4096)print(f"安全响应: {response.decode()}")except ssl.SSLError as e:print(f"SSL错误: {e}")except Exception as e:print(f"错误: {e}")finally:if 'secure_sock' in locals():secure_sock.close()

5.3、 性能优化

import select
import threadingdef non_blocking_socket_example():"""非阻塞Socket示例"""sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)sock.setblocking(False)  # 设置为非阻塞try:sock.connect(('example.com', 80))except BlockingIOError:# 非阻塞模式下连接会立即返回pass# 使用select等待socket可写ready = select.select([], [sock], [], 5)  # 5秒超时if ready[1]:# socket可写，连接建立成功sock.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')# 等待数据可读ready = select.select([sock], [], [], 5)if ready[0]:data = sock.recv(4096)print(f"非阻塞接收: {data.decode()}")else:print("连接超时")sock.close()

5.4、 资源管理

from contextlib import contextmanager@contextmanager
def socket_context(host, port, socket_type=socket.SOCK_STREAM):"""Socket上下文管理器"""sock = Nonetry:sock = socket.socket(socket.AF_INET, socket_type)yield sockexcept Exception as e:print(f"Socket错误: {e}")finally:if sock:sock.close()# 使用上下文管理器
def safe_socket_usage():with socket_context('example.com', 80) as sock:sock.connect(('example.com', 80))sock.send(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')response = sock.recv(4096)print(response.decode())

6、关键注意事项总结

  1. 异常处理: 总是处理网络异常和超时
  2. 资源清理: 使用try-finally或上下文管理器确保资源释放
  3. 编码问题: 注意字节串和字符串的转换
  4. 缓冲区管理: 正确处理数据分片和粘包问题
  5. 并发安全: 在多线程环境中注意线程安全
  6. 网络安全: 验证证书，使用加密连接
  7. 性能考虑: 使用异步编程处理高并发
  8. 协议兼容: 遵循网络协议规范
http://www.dtcms.com/a/554200.html

相关文章：

  • 好文与笔记分享 A Survey of Context Engineering for Large Language Models（中）
  • 英文网站推广网站前端程序制作开发
  • 寻找在山西运城专业做网站推广的中关村网站建设的公司
  • 微前端架构深度解析：从概念到落地的完整指南
  • 中国电力建设集团网站群做网站jsp好还是
  • 如何创建一个简单的网页南京企业网站做优化
  • 黑马JAVA+AI 加强09-2 IO流-文件字节输入流-文件字节输出流-文件字符流-文件字符输出流
  • Parasoft C/C++test如何在ARM DS-5环境中进行测试（下）
  • 佛山销售型网站建设重庆网红
  • Linux基础 -- 零拷贝之 splice
  • Go 协程
  • 做网站时怎样图片上传怎么才能让图片不变形有什么插件吗淄博住房和城乡建设局网站
  • leetcode1312.让字符串成为回文串的最少插入次数
  • 宜春做网站 黑酷seo快递网站建站需要什么
  • org.apache.commons.lang3都有什么常用的类
  • edas会议投稿显示格式错误+消除浮动块下面的空白
  • 宁波建设网站公司北京seo案例
  • 虚拟网站仿制教程河南国控建设集团招标网站
  • viewerjs+vue3 using typescript
  • U81904 【模板】树的直径
  • 如何将React自定义语法转化为标准JavaScript语法？
  • 自己做网站主机wordpress 引号被转义
  • 做营销网站推广快速开发安卓app
  • 文件基础操作详解
  • 【22】C语言 - 二维数组详解
  • 嵌入式项目代码架构与分层笔记
  • 自己房子做民宿挂什么网站数字今天科技 网站
  • 建设ca网站aws wordpress 集群
  • Rust数据类型（上）：标量类型全解析
  • BPC EPM表单常规设置