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

15.网络编程：让程序学会上网

news 2025/8/2 1:52:41

网络编程：让程序学会上网

🎯 前言：程序也要"冲浪"

想象一下，你的程序就像一个宅在家里的程序员（没错，就是你😄），整天对着屏幕，但是突然有一天，它想要走出家门，和外面的世界交流。这时候，网络编程就像是给程序安装了一张"网卡"，让它能够上网冲浪，和全世界的其他程序聊天、交换数据，甚至可以在网上"约会"！

今天我们就来学习如何让我们的Python程序变成一个"网络达人"，掌握发送HTTP请求、创建服务器、处理网络数据等技能。相信我，学完这篇文章，你的程序就不再是一个"网络小白"了！

📚 目录

网络编程基础概念
HTTP请求：程序的网络通信语言
创建简单的Web服务器
Socket编程：底层网络通信
实战项目：天气查询机器人
网络编程最佳实践
常见问题与解决方案

🧠 基础概念：网络编程是什么？

什么是网络编程？

网络编程就像是教程序说"网络语言"。就像你学英语是为了和外国人交流一样，程序学网络编程是为了和其他程序、服务器、网站交流。

网络通信的基本原理

想象网络通信就像寄信：

IP地址：就像邮政地址，告诉邮递员把信送到哪里
端口：就像门牌号，告诉邮递员敲哪扇门
协议：就像信封格式，大家都要遵守统一的规则

Python网络编程的常用库

# 内置库
import urllib.request    # 发送HTTP请求的老大哥
import urllib.parse      # URL处理专家
import socket           # 底层网络通信大师
import http.server      # 简易服务器搭建器# 第三方库
import requests         # HTTP请求的瑞士军刀（需要安装）
import flask            # Web框架小巧玲珑

📡 HTTP请求：程序的网络通信语言

使用urllib发送GET请求

import urllib.request
import jsondef get_weather_urllib(city):"""使用urllib获取天气信息就像程序员发微信问："今天天气怎么样？""""# 构造请求URL（假设我们有一个天气API）url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=your_api_key"try:# 发送请求，就像敲门with urllib.request.urlopen(url) as response:# 读取响应，就像听门里的人回答data = response.read().decode('utf-8')weather_data = json.loads(data)print(f"🌤️  {city}今天{weather_data['weather'][0]['description']}")print(f"🌡️  温度：{weather_data['main']['temp']}K")except Exception as e:print(f"😅 获取天气失败：{e}")# 使用示例
get_weather_urllib("Beijing")

使用requests库（推荐）

import requestsdef get_joke():"""获取一个冷笑话让程序也能讲笑话，比程序员的笑话还冷！"""try:# requests就像一个贴心的助手response = requests.get("https://api.jokes.one/jod")if response.status_code == 200:joke_data = response.json()joke = joke_data['contents']['jokes'][0]['joke']['text']print(f"😄 今日笑话：{joke}")else:print(f"😅 获取笑话失败，状态码：{response.status_code}")except Exception as e:print(f"😱 出错了：{e}")# 调用函数
get_joke()

POST请求：向服务器发送数据

import requestsdef send_feedback(name, message):"""发送用户反馈就像给网站留言："你们的网站不错，但是能不能加个夜间模式？""""url = "https://httpbin.org/post"  # 这是一个测试API# 准备要发送的数据data = {"name": name,"message": message,"timestamp": "2024-01-01"}# 设置请求头headers = {"Content-Type": "application/json","User-Agent": "Python-Student-Bot/1.0"}try:response = requests.post(url, json=data, headers=headers)if response.status_code == 200:print("✅ 反馈发送成功！")print(f"📝 服务器回应：{response.json()}")else:print(f"❌ 发送失败，状态码：{response.status_code}")except Exception as e:print(f"😱 网络错误：{e}")# 使用示例
send_feedback("小明", "你们的API很好用，给个赞！👍")

🖥️ Web服务器：让程序变身服务员

创建简单的HTTP服务器

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
from urllib.parse import urlparse, parse_qsclass MyHandler(BaseHTTPRequestHandler):"""自定义请求处理器就像一个礼貌的服务员，处理客人的各种要求"""def do_GET(self):"""处理GET请求"""# 解析URLparsed_url = urlparse(self.path)path = parsed_url.pathquery_params = parse_qs(parsed_url.query)if path == "/":self.send_homepage()elif path == "/hello":name = query_params.get('name', ['匿名'])[0]self.send_hello(name)elif path == "/joke":self.send_joke()else:self.send_404()def send_homepage(self):"""发送首页"""html = """<!DOCTYPE html><html><head><title>我的第一个Python服务器</title><meta charset="utf-8"></head><body><h1>🎉 欢迎来到我的服务器！</h1><p>这是用Python搭建的服务器，是不是很酷？</p><ul><li><a href="/hello?name=小明">打个招呼</a></li><li><a href="/joke">听个笑话</a></li></ul></body></html>"""self.send_response(200)self.send_header('Content-type', 'text/html; charset=utf-8')self.end_headers()self.wfile.write(html.encode('utf-8'))def send_hello(self, name):"""发送问候"""message = f"👋 你好，{name}！欢迎访问我的服务器！"self.send_json_response({"message": message})def send_joke(self):"""发送笑话"""jokes = ["为什么程序员总是搞混万圣节和圣诞节？因为 Oct 31 == Dec 25！","程序员的三大美德：懒惰、急躁和傲慢。","世界上有10种人：懂二进制的和不懂二进制的。"]import randomjoke = random.choice(jokes)self.send_json_response({"joke": joke})def send_404(self):"""发送404错误"""self.send_response(404)self.send_header('Content-type', 'text/html; charset=utf-8')self.end_headers()html = "<h1>404 - 页面不存在</h1><p>你要找的页面跑路了！😅</p>"self.wfile.write(html.encode('utf-8'))def send_json_response(self, data):"""发送JSON响应"""self.send_response(200)self.send_header('Content-type', 'application/json; charset=utf-8')self.end_headers()self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))def start_server():"""启动服务器"""server_address = ('localhost', 8000)httpd = HTTPServer(server_address, MyHandler)print("🚀 服务器启动成功！")print("📍 访问地址：http://localhost:8000")print("🛑 按Ctrl+C停止服务器")try:httpd.serve_forever()except KeyboardInterrupt:print("\n👋 服务器已停止")httpd.server_close()# 启动服务器
if __name__ == "__main__":start_server()

🔌 Socket编程：底层网络通信

创建TCP服务器

import socket
import threadingclass ChatServer:"""简易聊天服务器就像一个聊天室的管理员，负责转发消息"""def __init__(self, host='localhost', port=9999):self.host = hostself.port = portself.clients = []self.server_socket = Nonedef 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, addr = self.server_socket.accept()print(f"🎉 新用户连接：{addr}")self.clients.append(client_socket)# 为每个客户端创建一个线程client_thread = threading.Thread(target=self.handle_client,args=(client_socket, addr))client_thread.daemon = Trueclient_thread.start()except Exception as e:print(f"😱 服务器错误：{e}")finally:self.close()def handle_client(self, client_socket, addr):"""处理客户端消息"""try:while True:message = client_socket.recv(1024).decode('utf-8')if not message:breakprint(f"📝 {addr}: {message}")# 转发消息给所有客户端self.broadcast(f"{addr}: {message}", client_socket)except Exception as e:print(f"😅 客户端错误：{e}")finally:if client_socket in self.clients:self.clients.remove(client_socket)client_socket.close()print(f"👋 用户离开：{addr}")def broadcast(self, message, sender_socket):"""广播消息"""for client in self.clients:if client != sender_socket:try:client.send(message.encode('utf-8'))except:# 如果发送失败，移除客户端if client in self.clients:self.clients.remove(client)def close(self):"""关闭服务器"""if self.server_socket:self.server_socket.close()for client in self.clients:client.close()# 启动服务器
if __name__ == "__main__":server = ChatServer()try:server.start()except KeyboardInterrupt:print("\n🛑 服务器关闭")server.close()

创建TCP客户端

import socket
import threadingclass ChatClient:"""简易聊天客户端就像一个聊天应用，可以发送和接收消息"""def __init__(self, host='localhost', port=9999):self.host = hostself.port = portself.socket = Noneself.connected = Falsedef connect(self):"""连接到服务器"""try:self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.socket.connect((self.host, self.port))self.connected = Trueprint(f"✅ 连接成功：{self.host}:{self.port}")# 启动接收消息的线程receive_thread = threading.Thread(target=self.receive_messages)receive_thread.daemon = Truereceive_thread.start()# 开始发送消息self.send_messages()except Exception as e:print(f"😱 连接失败：{e}")def receive_messages(self):"""接收消息"""while self.connected:try:message = self.socket.recv(1024).decode('utf-8')if message:print(f"📨 {message}")else:breakexcept:breakdef send_messages(self):"""发送消息"""print("💬 开始聊天！输入'quit'退出")while self.connected:try:message = input()if message.lower() == 'quit':breakself.socket.send(message.encode('utf-8'))except:breakself.disconnect()def disconnect(self):"""断开连接"""self.connected = Falseif self.socket:self.socket.close()print("👋 已断开连接")# 启动客户端
if __name__ == "__main__":client = ChatClient()client.connect()

🌤️ 实战项目：天气查询机器人

import requests
import json
from datetime import datetimeclass WeatherBot:"""天气查询机器人就像一个贴心的天气预报员，随时为你播报天气"""def __init__(self, api_key):self.api_key = api_keyself.base_url = "https://api.openweathermap.org/data/2.5/weather"def get_weather(self, city):"""获取天气信息"""try:params = {'q': city,'appid': self.api_key,'units': 'metric',  # 使用摄氏度'lang': 'zh_cn'     # 中文描述}response = requests.get(self.base_url, params=params)if response.status_code == 200:data = response.json()return self.format_weather_info(data)else:return f"😅 获取{city}天气失败，请检查城市名称"except Exception as e:return f"😱 网络错误：{e}"def format_weather_info(self, data):"""格式化天气信息"""city = data['name']country = data['sys']['country']temp = data['main']['temp']feels_like = data['main']['feels_like']humidity = data['main']['humidity']pressure = data['main']['pressure']description = data['weather'][0]['description']# 根据天气选择emojiweather_emoji = self.get_weather_emoji(data['weather'][0]['main'])weather_info = f"""
🌍 {city}, {country}
{weather_emoji} {description}
🌡️  温度：{temp}°C (体感：{feels_like}°C)
💧 湿度：{humidity}%
📊 气压：{pressure}hPa
⏰ 查询时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"""return weather_info.strip()def get_weather_emoji(self, weather_main):"""根据天气类型返回emoji"""emoji_map = {'Clear': '☀️','Clouds': '☁️','Rain': '🌧️','Drizzle': '🌦️','Thunderstorm': '⛈️','Snow': '❄️','Mist': '🌫️','Fog': '🌫️','Haze': '🌫️'}return emoji_map.get(weather_main, '🌤️')def get_multiple_cities_weather(self, cities):"""获取多个城市的天气"""results = []for city in cities:weather_info = self.get_weather(city)results.append(f"📍 {city.upper()}\n{weather_info}")return "\n" + "="*50 + "\n".join(results)# 使用示例
def main():# 注意：你需要从 https://openweathermap.org/ 申请免费的API密钥API_KEY = "your_api_key_here"if API_KEY == "your_api_key_here":print("⚠️  请先申请OpenWeatherMap API密钥并替换API_KEY")returnbot = WeatherBot(API_KEY)print("🤖 天气查询机器人启动！")print("💡 输入城市名称查询天气，输入'quit'退出")while True:city = input("\n🏙️  请输入城市名称：").strip()if city.lower() == 'quit':print("👋 再见！")breakif not city:print("😅 请输入有效的城市名称")continueprint("⏳ 正在查询天气...")weather_info = bot.get_weather(city)print(weather_info)if __name__ == "__main__":main()

🚀 进阶技巧：让你的网络程序更专业

1. 异步网络编程

import asyncio
import aiohttpasync def fetch_url(session, url):"""异步获取URL内容"""try:async with session.get(url) as response:return await response.text()except Exception as e:return f"Error: {e}"async def fetch_multiple_urls(urls):"""同时获取多个URL"""async with aiohttp.ClientSession() as session:tasks = [fetch_url(session, url) for url in urls]results = await asyncio.gather(*tasks)return results# 使用示例
async def main():urls = ["https://httpbin.org/delay/1","https://httpbin.org/delay/2","https://httpbin.org/delay/3"]print("⏳ 开始异步请求...")start_time = asyncio.get_event_loop().time()results = await fetch_multiple_urls(urls)end_time = asyncio.get_event_loop().time()print(f"✅ 完成！用时：{end_time - start_time:.2f}秒")for i, result in enumerate(results):print(f"URL {i+1}: {len(result)} 字符")# 运行异步程序
# asyncio.run(main())

2. 网络请求的错误处理和重试

import requests
import time
from functools import wrapsdef retry(max_attempts=3, delay=1):"""重试装饰器"""def decorator(func):@wraps(func)def wrapper(*args, **kwargs):for attempt in range(max_attempts):try:return func(*args, **kwargs)except Exception as e:if attempt == max_attempts - 1:raise eprint(f"😅 第{attempt + 1}次尝试失败：{e}")time.sleep(delay)return Nonereturn wrapperreturn decorator@retry(max_attempts=3, delay=2)
def reliable_request(url):"""可靠的网络请求"""response = requests.get(url, timeout=10)response.raise_for_status()  # 如果状态码不是200，抛出异常return response.json()# 使用示例
try:data = reliable_request("https://api.github.com/users/octocat")print(f"✅ 获取用户信息成功：{data['name']}")
except Exception as e:print(f"😱 最终失败：{e}")

3. 网络爬虫的礼貌做法

import requests
from bs4 import BeautifulSoup
import time
import randomclass PoliteCrawler:"""礼貌的网络爬虫就像一个有礼貌的访客，不会给服务器造成压力"""def __init__(self, delay_range=(1, 3)):self.delay_range = delay_rangeself.session = requests.Session()self.session.headers.update({'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'})def crawl_with_delay(self, url):"""带延迟的爬取"""# 随机延迟，避免给服务器造成压力delay = random.uniform(*self.delay_range)time.sleep(delay)try:response = self.session.get(url, timeout=10)response.raise_for_status()return response.textexcept Exception as e:print(f"😅 爬取失败：{e}")return Nonedef parse_news_titles(self, html):"""解析新闻标题"""if not html:return []soup = BeautifulSoup(html, 'html.parser')# 这里的选择器需要根据实际网站调整titles = soup.find_all('h3', class_='news-title')return [title.get_text().strip() for title in titles]# 使用示例
crawler = PoliteCrawler()
html = crawler.crawl_with_delay("https://example.com/news")
titles = crawler.parse_news_titles(html)
print(f"📰 找到{len(titles)}条新闻标题")

🔧 常见问题与解决方案

Q1: 网络请求超时怎么办？

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retrydef create_robust_session():"""创建一个健壮的请求会话"""session = requests.Session()# 设置重试策略retry_strategy = Retry(total=3,backoff_factor=1,status_forcelist=[429, 500, 502, 503, 504])adapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)return session# 使用示例
session = create_robust_session()
try:response = session.get("https://httpbin.org/delay/5", timeout=10)print("✅ 请求成功！")
except requests.exceptions.RequestException as e:print(f"😱 请求失败：{e}")

Q2: 如何处理不同编码的网页？

import requests
import chardetdef smart_decode(response):"""智能解码网页内容"""# 首先尝试response自带的编码if response.encoding != 'ISO-8859-1':return response.text# 如果是默认编码，尝试检测真实编码raw_data = response.contentdetected = chardet.detect(raw_data)if detected['confidence'] > 0.8:return raw_data.decode(detected['encoding'])else:# 尝试常见编码for encoding in ['utf-8', 'gbk', 'gb2312']:try:return raw_data.decode(encoding)except UnicodeDecodeError:continue# 如果都失败了，使用错误处理return raw_data.decode('utf-8', errors='ignore')# 使用示例
response = requests.get("https://example.com")
content = smart_decode(response)
print(f"📄 成功解码网页内容：{len(content)}个字符")

Q3: 如何优雅地关闭网络连接？

import socket
import signal
import sysclass GracefulServer:"""优雅关闭的服务器"""def __init__(self, host='localhost', port=8888):self.host = hostself.port = portself.server_socket = Noneself.running = Falsedef signal_handler(self, signum, frame):"""处理关闭信号"""print(f"\n🛑 收到关闭信号：{signum}")self.stop()def start(self):"""启动服务器"""# 注册信号处理器signal.signal(signal.SIGINT, self.signal_handler)signal.signal(signal.SIGTERM, self.signal_handler)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)self.running = Trueprint(f"🚀 服务器启动：{self.host}:{self.port}")while self.running:try:client_socket, addr = self.server_socket.accept()# 处理客户端连接self.handle_client(client_socket, addr)except OSError:# 服务器关闭时会抛出异常breakfinally:self.cleanup()def handle_client(self, client_socket, addr):"""处理客户端"""try:data = client_socket.recv(1024)response = b"HTTP/1.1 200 OK\r\n\r\nHello World!"client_socket.send(response)finally:client_socket.close()def stop(self):"""停止服务器"""print("⏳ 正在关闭服务器...")self.running = Falseif self.server_socket:self.server_socket.close()def cleanup(self):"""清理资源"""print("🧹 清理资源完成")sys.exit(0)# 使用示例
if __name__ == "__main__":server = GracefulServer()server.start()