用 curl 和 Python 从零开始写 HTTP 请求与响应处理实战指南及常见连接超时错误排查
一、先搞清楚 HTTP 到底是什么
HTTP 全称是超文本传输协议,你可以把它理解为浏览器和服务器之间的一种”约定好的对话方式”。当你输入一个网址,浏览器会向服务器发送一段文本(请求),服务器收到后回复另一段文本(响应),你看到页面就是这来回几次”对话”的结果。
这背后其实就三样东西:请求方法、请求头、请求体,以及对应的状态码、响应头、响应体。下面咱们用两个最实用的工具——curl 和 Python 的 requests 库——手把手演示怎么发送请求、处理响应,以及遇到超时错误该怎么办。
二、curl 命令行工具的实战用法
curl 是 Linux、macOS 自带的命令行工具,Windows 10 以上版本也内置了。它不需要安装任何库,一条命令就能发 HTTP 请求,调试接口非常方便。
2.1 最基础的 GET 请求
# 最简单的请求,只看响应体
curl https://httpbin.org/get
# 加上 -v 参数,把请求头和响应头都打印出来,调试时非常有用
curl -v https://httpbin.org/get
输出大概长这样:
* Trying 99.86.38.121...
* TCP connection established
* TLS handshake completed
> GET /get HTTP/1.1
> Host: httpbin.org
> User-Agent: curl/8.1.2
> Accept: */*
< HTTP/1.1 200 OK
< Content-Type: application/json
< Content-Length: 345
<
{
"args": {},
"headers": {
"Accept": "*/*",
"Host": "httpbin.org",
"User-Agent": "curl/8.1.2"
},
"origin": "1.2.3.4",
"url": "https://httpbin.org/get"
}
这里要注意 HTTP/1.1 200 OK,200 就是状态码,表示请求成功了。Content-Type: application/json 告诉客户端响应内容是 JSON 格式。
2.2 发送带参数的 GET 请求
# 查询参数直接拼在 URL 后面
curl "https://httpbin.org/get?name=张三&age=25"
# 用 -G 和 -d 参数组合,curl 会自动处理编码
curl -G -d "name=张三" -d "age=25" https://httpbin.org/get
2.3 POST 请求与 JSON 数据
# 发送 JSON 数据的 POST 请求
curl -X POST https://httpbin.org/post \
-H "Content-Type: application/json" \
-d '{"username": "ag nes", "password": "123456"}'
# 发送表单数据(类似网页表单提交)
curl -X POST https://httpbin.org/post \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "username=agnes&password=123456"
参数解释:
-X POST:指定请求方法为 POST-H:添加请求头-d:发送请求体数据
2.4 查看响应头信息
# 只打印响应头,不打印响应体
curl -I https://httpbin.org/get
# 或者用 --head 别名
curl --head https://httpbin.org/get
输出:
HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 345
Connection: keep-alive
Server: nginx
Date: Mon, 15 Jul 2024 10:00:00 GMT
2.5 处理 Cookie 和会话
# 保存服务器返回的 Cookie 到文件
curl -c cookies.txt https://httpbin.org/cookies/set/name/agnes
# 发送请求时带上之前的 Cookie
curl -b cookies.txt https://httpbin.org/cookies
2.6 设置超时参数(关键!)
# 设置连接超时时间为 5 秒
curl --connect-timeout 5 https://httpbin.org/get
# 设置请求整体超时时间为 10 秒
curl --max-time 10 https://httpbin.org/get
# 两个都设置,推荐这样做
curl --connect-timeout 5 --max-time 10 https://httpbin.org/get
超时参数说明:
--connect-timeout:连接服务器的最大时间,超过这个时间还没连上就报错--max-time:整个请求(包括下载数据)的最大时间
2.7 下载文件
# 下载到指定文件
curl -o report.pdf https://example.com/files/report.pdf
# 下载时显示进度条
curl -# -o report.pdf https://example.com/files/report.pdf
# 断点续传(大文件很有用)
curl -C - -o report.pdf https://example.com/files/report.pdf
2.8 处理 HTTPS 证书问题
# 跳过 SSL 证书验证(测试环境用,生产环境不要这样!)
curl -k https://self-signed.badssl.com/
# 指定 CA 证书文件
curl --cacert ca-bundle.crt https://example.com
三、Python requests 库的完整使用指南
Python 的 requests 库是发 HTTP 请求最常用的工具,简洁优雅,比内置的 urllib 好用得多。
3.1 安装
pip install requests
3.2 GET 请求基础用法
import requests
# 最简单的 GET 请求
response = requests.get("https://httpbin.org/get")
# 查看状态码
print(response.status_code) # 输出: 200
# 查看响应内容(自动解析为 Python 对象)
print(response.json())
# 查看响应文本
print(response.text)
# 查看响应头
print(response.headers)
# 查看原始字节
print(response.content)
3.3 带参数的 GET 请求
import requests
# 方式一:直接拼在 URL 里(不推荐,参数容易出错)
response = requests.get("https://httpbin.org/get?name=agnes&age=25")
# 方式二:使用 params 参数(推荐,自动处理编码)
params = {
"name": "agnes",
"age": 25,
"city": "北京"
}
response = requests.get("https://httpbin.org/get", params=params)
print(response.url)
# 输出: https://httpbin.org/get?name=agnes&age=25&city=%E5%8C%97%E4%BA%AC
3.4 POST 请求与 JSON 数据
import requests
# 发送 JSON 数据
url = "https://httpbin.org/post"
data = {
"username": "agnes",
"password": "123456",
"role": "admin"
}
# 方式一:手动序列化(不推荐)
response = requests.post(url, json=data)
# 方式二:自动序列化为 JSON,并设置 Content-Type
response = requests.post(url, json=data)
print(response.status_code) # 200
print(response.json()) # 返回的 JSON 数据
注意 json=data 参数,它会自动:
- 将 Python 字典序列化为 JSON 字符串
- 设置请求头
Content-Type: application/json
如果手动设置:
import json
response = requests.post(
url,
data=json.dumps(data),
headers={"Content-Type": "application/json"}
)
3.5 发送表单数据
import requests
url = "https://httpbin.org/post"
form_data = {
"username": "agnes",
"password": "123456"
}
# 发送表单数据(自动设置 Content-Type: application/x-www-form-urlencoded)
response = requests.post(url, data=form_data)
print(response.json())
3.6 上传文件
import requests
url = "https://httpbin.org/post"
# 上传单个文件
with open("report.pdf", "rb") as f:
files = {"file": ("report.pdf", f, "application/pdf")}
response = requests.post(url, files=files)
# 上传多个文件
with open("photo1.jpg", "rb") as f1, open("photo2.jpg", "rb") as f2:
files = {
"photo1": ("photo1.jpg", f1, "image/jpeg"),
"photo2": ("photo2.jpg", f2, "image/jpeg"),
}
response = requests.post(url, files=files)
print(response.json())
3.7 自定义请求头
import requests
url = "https://httpbin.org/headers"
headers = {
"User-Agent": "MyApp/1.0",
"Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
"Accept": "application/json",
"X-Custom-Header": "custom-value"
}
response = requests.get(url, headers=headers)
print(response.json())
3.8 处理 Cookie 和会话
import requests
# 方式一:手动处理 Cookie
headers = {"Cookie": "session_id=abc123"}
response = requests.get("https://httpbin.org/cookies", headers=headers)
# 方式二:使用 Session 对象(推荐,自动管理 Cookie)
session = requests.Session()
# 第一次请求,服务器会设置 Cookie
session.get("https://httpbin.org/cookies/set/name/agnes")
# 后续请求自动带上 Cookie
response = session.get("https://httpbin.org/cookies")
print(response.json())
# {'cookies': {'name': 'agnes'}}
3.9 处理重定向
import requests
# 默认情况下,requests 会自动跟随重定向(最多 30 次)
response = requests.get("https://httpbin.org/redirect/3")
print(response.status_code) # 200
print(response.url) # 最终跳转后的 URL
# 禁止自动跟随重定向
response = requests.get(
"https://httpbin.org/redirect/3",
allow_redirects=False
)
print(response.status_code) # 302
print(response.headers["Location"]) # 重定向目标 URL
3.10 超时设置(重点!)
import requests
# 方式一:单个超时参数
response = requests.get(
"https://httpbin.org/delay/3",
timeout=10 # 等待 10 秒
)
# 方式二:分别设置连接超时和读取超时(推荐)
try:
response = requests.get(
"https://httpbin.org/get",
timeout=(5, 10) # (连接超时, 读取超时)
)
except requests.exceptions.Timeout:
print("请求超时了")
except requests.exceptions.ConnectionError:
print("连接错误")
超时参数说明:
timeout=(5, 10):前一个数字是连接超时(建立连接的最大时间),后一个数字是读取超时(等待服务器响应的最大时间)- 如果只传一个数字,两个超时都用这个值
四、响应处理的完整技巧
4.1 检查状态码
import requests
response = requests.get("https://httpbin.org/status/404")
# 方式一:手动检查
if response.status_code == 200:
print("成功")
elif response.status_code == 404:
print("找不到页面")
elif response.status_code == 500:
print("服务器错误")
# 方式二:使用 raise_for_status()(推荐用于 API 调用)
try:
response.raise_for_status()
except requests.exceptions.HTTPError as e:
print(f"HTTP 错误: {e}")
# 404 Client Error: Not Found
4.2 解析 JSON 响应
import requests
response = requests.get("https://httpbin.org/get")
# 方式一:使用 json() 方法
data = response.json()
print(data["headers"]["Host"])
# 方式二:手动解析
import json
data = json.loads(response.text)
# 处理可能的 JSON 解析错误
try:
data = response.json()
except requests.exceptions.JSONDecodeError:
print("响应不是有效的 JSON")
print(response.text)
4.3 处理二进制响应(图片、PDF 等)
import requests
# 下载图片
response = requests.get("https://httpbin.org/image/png")
with open("downloaded_image.png", "wb") as f:
f.write(response.content)
# 流式下载大文件(内存友好)
response = requests.get(
"https://httpbin.org/stream/100",
stream=True
)
with open("large_file.bin", "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
4.4 使用响应对象的全部属性
import requests
response = requests.get("https://httpbin.org/get")
print(f"状态码: {response.status_code}")
print(f"请求 URL: {response.url}")
print(f"请求历史: {response.history}") # 重定向历史
print(f"响应头: {response.headers}")
print(f"最后一条响应头: {response.headers['Content-Type']}")
print(f"编码: {response.encoding}")
print(f"内容长度: {len(response.content)} 字节")
print(f"执行时间: {response.elapsed} 秒")
print(f"请求方法: {response.request.method}")
五、连接超时错误排查完全指南
超时错误是最常见的网络问题之一,下面从浅入深讲解。
5.1 常见的超时错误类型
import requests
from requests.exceptions import (
Timeout,
ConnectionError,
HTTPError,
TooManyRedirects,
RequestException
)
# 各种异常对应不同情况
try:
response = requests.get("https://httpbin.org/get", timeout=3)
except Timeout:
print("请求超时了") # 连接超时或读取超时
except ConnectionError:
print("网络连接错误") # DNS 失败、拒绝连接等
except HTTPError:
print("HTTP 错误") # 4xx 或 5xx 状态码
except TooManyRedirects:
print("重定向次数过多") # 超过 30 次重定向
except RequestException:
print("请求异常") # 其他所有异常
5.2 连接超时 vs 读取超时
连接超时 (Connect Timeout) 读取超时 (Read Timeout)
| |
v v
建立 TCP 连接 等待服务器响应数据
比如 DNS 解析 + TCP 握手 比如服务器处理请求
超过这个时间就报错 超过这个时间就报错
import requests
# 连接超时 5 秒,读取超时 10 秒
response = requests.get(
"https://httpbin.org/get",
timeout=(5, 10)
)
# 只设置读取超时(连接不超时)
response = requests.get(
"https://httpbin.org/get",
timeout=10 # 相当于 (None, 10)
)
5.3 常见超时原因和解决方案
原因一:服务器响应慢
import requests
import time
# 测试服务器响应时间
start = time.time()
response = requests.get(
"https://httpbin.org/delay/3", # 故意延迟 3 秒
timeout=10
)
elapsed = time.time() - start
print(f"请求耗时: {elapsed:.2f} 秒")
原因二:DNS 解析失败
import requests
import socket
# 测试 DNS 解析
try:
socket.getaddrinfo("example.com", 443)
print("DNS 解析成功")
except socket.gaierror as e:
print(f"DNS 解析失败: {e}")
# requests 中捕获 DNS 错误
try:
response = requests.get("https://invalid-domain-12345.com", timeout=5)
except requests.exceptions.ConnectionError as e:
print(f"连接错误: {e}")
原因三:防火墙或网络限制
import requests
# 代理设置
proxies = {
"http": "http://127.0.0.1:7890",
"https": "http://127.0.0.1:7890"
}
response = requests.get(
"https://httpbin.org/get",
proxies=proxies,
timeout=10
)
# 使用 SOCKS 代理
proxies = {
"http": "socks5://127.0.0.1:1080",
"https": "socks5://127.0.0.1:1080"
}
原因四:TLS/SSL 握手失败
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 自定义 SSL 配置
session = requests.Session()
# 添加重试机制
retry_strategy = Retry(
total=3, # 最大重试次数
backoff_factor=1, # 重试间隔:1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504] # 对这些状态码重试
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
response = session.get("https://httpbin.org/get", timeout=10)
5.4 完整的超时处理模板
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HTTPClient:
"""带超时和重试机制的 HTTP 客户端"""
def __init__(self, connect_timeout=5, read_timeout=10, max_retries=3):
self.connect_timeout = connect_timeout
self.read_timeout = read_timeout
self.max_retries = max_retries
self.session = requests.Session()
self._setup_retry()
def _setup_retry(self):
"""配置重试策略"""
retry_strategy = Retry(
total=self.max_retries,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST", "PUT", "DELETE"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
def get(self, url, params=None, headers=None):
"""发送 GET 请求"""
timeout = (self.connect_timeout, self.read_timeout)
try:
response = self.session.get(
url,
params=params,
headers=headers,
timeout=timeout
)
response.raise_for_status()
return response
except requests.exceptions.Timeout:
logger.error(f"请求超时: {url}")
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"连接错误: {url} - {e}")
raise
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP 错误: {url} - {e}")
raise
def post(self, url, json=None, data=None, headers=None):
"""发送 POST 请求"""
timeout = (self.connect_timeout, self.read_timeout)
try:
response = self.session.post(
url,
json=json,
data=data,
headers=headers,
timeout=timeout
)
response.raise_for_status()
return response
except requests.exceptions.Timeout:
logger.error(f"请求超时: {url}")
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"连接错误: {url} - {e}")
raise
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP 错误: {url} - {e}")
raise
def close(self):
"""关闭会话"""
self.session.close()
# 使用示例
if __name__ == "__main__":
client = HTTPClient(connect_timeout=5, read_timeout=10, max_retries=3)
try:
# GET 请求
response = client.get("https://httpbin.org/get", params={"key": "value"})
print(response.json())
# POST 请求
response = client.post(
"https://httpbin.org/post",
json={"username": "agnes", "password": "123456"}
)
print(response.json())
finally:
client.close()
5.5 调试工具:启用请求日志
import requests
import logging
# 启用 requests 的调试日志
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
# 现在所有请求和响应都会打印出来
response = requests.get("https://httpbin.org/get", timeout=10)
5.6 诊断超时问题的步骤
import socket
import requests
import time
def diagnose_connection(url, timeout=5):
"""诊断网络连接问题"""
results = {}
# 1. DNS 解析
print("【1】DNS 解析测试")
start = time.time()
try:
socket.getaddrinfo(url.split("//")[1].split("/")[0], 443)
elapsed = time.time() - start
results["dns"] = f"成功 ({elapsed:.3f}s)"
print(f" ✓ DNS 解析成功,耗时 {elapsed:.3f}s")
except socket.gaierror as e:
results["dns"] = f"失败: {e}"
print(f" ✗ DNS 解析失败: {e}")
return results
# 2. TCP 连接
print("【2】TCP 连接测试")
hostname = url.split("//")[1].split("/")[0]
port = 443 if url.startswith("https") else 80
start = time.time()
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
sock.connect((hostname, port))
sock.close()
elapsed = time.time() - start
results["tcp"] = f"成功 ({elapsed:.3f}s)"
print(f" ✓ TCP 连接成功,耗时 {elapsed:.3f}s")
except socket.timeout:
results["tcp"] = "超时"
print(f" ✗ TCP 连接超时")
except ConnectionRefusedError:
results["tcp"] = "被拒绝"
print(f" ✗ TCP 连接被拒绝")
except Exception as e:
results["tcp"] = f"失败: {e}"
print(f" ✗ TCP 连接失败: {e}")
# 3. HTTP 请求
print("【3】HTTP 请求测试")
start = time.time()
try:
response = requests.get(url, timeout=(timeout, timeout * 2))
elapsed = time.time() - start
results["http"] = f"成功 (状态码: {response.status_code}, 耗时: {elapsed:.3f}s)"
print(f" ✓ HTTP 请求成功,状态码: {response.status_code}")
except requests.exceptions.Timeout:
results["http"] = "超时"
print(f" ✗ HTTP 请求超时")
except requests.exceptions.ConnectionError as e:
results["http"] = f"连接错误: {e}"
print(f" ✗ HTTP 连接错误: {e}")
except Exception as e:
results["http"] = f"失败: {e}"
print(f" ✗ HTTP 请求失败: {e}")
return results
# 使用诊断工具
if __name__ == "__main__":
url = "https://httpbin.org/get"
results = diagnose_connection(url)
print("\n诊断结果汇总:")
for key, value in results.items():
print(f" {key}: {value}")
六、curl 和 Python 对比速查表
| 功能 | curl | Python requests |
|---|---|---|
| 安装 | 系统自带 | pip install requests |
| GET 请求 | curl url |
requests.get(url) |
| POST JSON | curl -X POST -H "Content-Type: application/json" -d '{"key":"value"}' url |
requests.post(url, json={"key": "value"}) |
| 超时设置 | --connect-timeout 5 --max-time 10 |
timeout=(5, 10) |
| 查看响应头 | curl -I url |
response.headers |
| 下载文件 | curl -o file url |
open("file", "wb").write(r.content) |
| 设置代理 | curl -x http://proxy:port url |
requests.get(url, proxies={"http": "..."}) |
| 发送 Cookie | curl -b "name=value" url |
requests.get(url, cookies={"name": "value"}) |
| 调试模式 | curl -v url |
启用 logging |
七、常见错误速查表
错误信息 原因 解决方法
─────────────────────────────────────────────────────────────────────
ConnectionTimeout 连接服务器超时 检查网络、增加超时时间
ReadTimeout 等待响应超时 服务器处理慢,增加超时时间
ConnectionError 网络连接失败 检查 URL、代理、防火墙
HTTPError (4xx) 客户端错误 检查请求参数、认证
HTTPError (5xx) 服务器错误 联系服务端开发者
SSLError SSL/TLS 握手失败 检查证书、更新 CA 包
TooManyRedirects 重定向循环 检查 URL 是否正确
八、一个完整的实战示例
假设你要写一个 API 客户端,需要处理各种异常情况:
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class APIClient:
"""一个健壮的 API 客户端,处理超时和重试"""
def __init__(
self,
base_url: str,
connect_timeout: float = 5.0,
read_timeout: float = 10.0,
max_retries: int = 3
):
self.base_url = base_url.rstrip("/")
self.timeout = (connect_timeout, read_timeout)
self.session = requests.Session()
# 配置重试策略
retry = Retry(
total=max_retries,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST", "PUT", "DELETE"]
)
adapter = HTTPAdapter(max_retries=retry)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
def get(self, endpoint: str, params: dict = None, headers: dict = None):
"""发送 GET 请求"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
logger.info(f"GET {url}")
response = self.session.get(url, params=params, headers=headers, timeout=self.timeout)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
logger.error(f"请求超时: {url}")
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"连接错误: {url} - {e}")
raise
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP 错误: {url} - {e}")
raise
def post(self, endpoint: str, json: dict = None, data: dict = None):
"""发送 POST 请求"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
logger.info(f"POST {url}")
response = self.session.post(url, json=json, data=data, timeout=self.timeout)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
logger.error(f"请求超时: {url}")
raise
except requests.exceptions.ConnectionError as e:
logger.error(f"连接错误: {url} - {e}")
raise
except requests.exceptions.HTTPError as e:
logger.error(f"HTTP 错误: {url} - {e}")
raise
def close(self):
"""关闭会话"""
self.session.close()
# 使用示例
if __name__ == "__main__":
client = APIClient(
base_url="https://httpbin.org",
connect_timeout=5,
read_timeout=10,
max_retries=3
)
try:
# 测试 GET
result = client.get("/get", params={"key": "value"})
print(f"GET 结果: {result}")
# 测试 POST
result = client.post("/post", json={"username": "agnes"})
print(f"POST 结果: {result}")
finally:
client.close()
希望这份指南能帮你彻底搞懂 HTTP 请求与响应的处理,遇到超时错误时也知道怎么排查。记住,超时设置不是可有可无的,生产代码里一定要设,否则你的程序可能会永远卡在网络请求上。
