Nginx 反向代理与负载均衡的实现原理
·
Nginx 反向代理与负载均衡的实现原理
一、反向代理的实现原理
1. 整体流程
客户端 Nginx 后端服务器
│ │ │
│ ① TCP 三次握手 │ │
│ ──────────────────────→│ │
│ │ │
│ ② 发送 HTTP 请求 │ │
│ ──────────────────────→│ │
│ │ ③ 解析请求，匹配 location │
│ │ ④ 选择后端（负载均衡算法） │
│ │ │
│ │ ⑤ 与后端建立 TCP 连接 │
│ │ ─────────────────────────→│
│ │ │
│ │ ⑥ 转发请求（修改 Header） │
│ │ ─────────────────────────→│
│ │ │
│ │ ⑦ 后端返回响应 │
│ │ ←─────────────────────────│
│ │ │
│ ⑧ Nginx 转发响应给客户端 │ │
│ ←──────────────────────│ │
│ │ │
2. Nginx 内部处理的 11 个阶段
Nginx 处理每个 HTTP 请求分为 11 个阶段，按顺序执行：
请求进入
│
▼
┌──────────────────────────────────────────────────────────┐
│ 1. POST_READ ── 读取请求内容后 │
│ 2. SERVER_REWRITE ── server 块中的 rewrite │
│ 3. FIND_CONFIG ── 匹配 location（核心！） │
│ 4. REWRITE ── location 中的 rewrite │
│ 5. POST_REWRITE ── rewrite 后的处理 │
│ 6. PREACCESS ── 访问控制预处理（limit_req 限流） │
│ 7. ACCESS ── 访问控制（allow/deny, auth） │
│ 8. POST_ACCESS ── 访问控制后处理 │
│ 9. PRECONTENT ── 内容生成前（try_files） │
│ 10. CONTENT ── 生成内容（proxy_pass 在这里！） │
│ 11. LOG ── 记录日志 │
└──────────────────────────────────────────────────────────┘
│
▼
响应返回客户端
反向代理发生在第 10 阶段（CONTENT），由 ngx_http_proxy_module 模块执行。
3. proxy_pass 的工作细节
location /api/ {
proxy_pass http://127.0.0.1:8000;
}
Nginx 在这一步做了什么？
步骤 1：URI 改写
请求: GET /api/users?page=1
转发: GET /api/users?page=1 → http://127.0.0.1:8000/api/users?page=1
# 如果 proxy_pass 带了路径：
location /api/ {
proxy_pass http://127.0.0.1:8000/v2/;
}
# /api/users → /v2/users（替换匹配到的前缀）
步骤 2：修改 HTTP Header
原始请求头：
Host: www.example.com
X-Real-IP: (无)
Connection: keep-alive
转发的请求头：
Host: 127.0.0.1:8000 ← 默认改为后端地址
Connection: close ← 默认改为 close
步骤 3：建立与后端的 TCP 连接（或复用连接池中的连接）
步骤 4：发送请求到后端，等待响应
步骤 5：接收后端响应，转发给客户端
4. 连接管理：两条独立的 TCP 连接
连接 A（客户端↔Nginx） 连接 B（Nginx↔后端）
│ │
客户端 ←│→ Nginx 的 socket Nginx 的 socket ←│→ 后端
│ (80/443端口) (随机端口) │ (8000端口)
│ │
长连接（keepalive） 可配置连接池
客户端控制 Nginx 控制
# 优化后端连接
upstream backend {
server 127.0.0.1:8000;
keepalive 32; # 保持 32 个空闲长连接到后端（连接池）
}
location /api/ {
proxy_pass http://backend;
proxy_http_version 1.1; # 使用 HTTP/1.1 支持长连接
proxy_set_header Connection ""; # 清除 Connection: close
}
连接池的作用：
没有连接池：每次请求都要 TCP 三次握手 → 四次挥手
请求1: [握手] → [发请求] → [收响应] → [挥手]
请求2: [握手] → [发请求] → [收响应] → [挥手] ← 重复开销
有连接池：复用已建立的连接
请求1: [握手] → [发请求] → [收响应] ← 连接保持
请求2: [发请求] → [收响应] ← 直接复用！
请求3: [发请求] → [收响应]
5. 缓冲机制（核心优化！）
没有缓冲：
慢客户端(3G网络) ←──── Nginx ←──── 后端（Gunicorn Worker）
慢速传输 Worker 被占用，直到客户端接收完毕
← Worker 数量有限，被白白浪费！
有缓冲（默认开启）：
慢客户端(3G网络) ←──── Nginx ←──── 后端（Gunicorn Worker）
慢速传输 快速传输
后端瞬间传完 → Worker 立即释放！
Nginx 把响应存在缓冲区，慢慢发给客户端
location /api/ {
proxy_pass http://backend;
# 缓冲配置
proxy_buffering on; # 开启缓冲（默认 on）
proxy_buffer_size 4k; # 响应头缓冲区
proxy_buffers 8 16k; # 响应体缓冲区（8 个，每个 16k）
proxy_busy_buffers_size 32k; # 忙时最大缓冲
proxy_temp_file_write_size 64k; # 超出内存缓冲时写入临时文件
}
二、负载均衡的实现原理
1. 核心数据结构
upstream backend_servers {
server 192.168.1.101:8000 weight=5;
server 192.168.1.102:8000 weight=3;
server 192.168.1.103:8000 weight=2;
}
Nginx 内部用一个数组存储这些后端：
┌────────────────────────────────────────────────────────┐
│ upstream 后端服务器列表 │
│ │
│ peers[] = [ │
│ { addr: 101, weight: 5, current_weight: 0, │
│ fails: 0, max_fails: 1, down: false }, │
│ { addr: 102, weight: 3, current_weight: 0, │
│ fails: 0, max_fails: 1, down: false }, │
│ { addr: 103, weight: 2, current_weight: 0, │
│ fails: 0, max_fails: 1, down: false }, │
│ ] │
│ total_weight = 10 │
└────────────────────────────────────────────────────────┘
2. 加权轮询算法（Weighted Round-Robin）— 默认算法，最重要！
Nginx 使用的是平滑加权轮询算法（Smooth Weighted Round-Robin），确保请求分布均匀，不会出现某个后端连续收到大量请求。
用 Python 实现：
class Server:
def __init__(self, name, weight):
self.name = name
self.weight = weight # 配置权重（固定不变）
self.current_weight = 0 # 当前权重（动态变化）
def __repr__(self):
return f"{self.name}(w={self.weight}, cw={self.current_weight})"
def smooth_weighted_round_robin(servers, num_requests):
"""
Nginx 的平滑加权轮询算法
每次选择 current_weight 最大的服务器，
选中后该服务器的 current_weight 减去 total_weight
"""
total_weight = sum(s.weight for s in servers)
print(f"服务器: {[(s.name, f'权重={s.weight}') for s in servers]}")
print(f"总权重: {total_weight}\n")
print(f"{'请求':>4} | {'选中前 current_weight':>30} | {'选中':>6} | {'选中后 current_weight':>30}")
print("-" * 90)
results = []
for i in range(num_requests):
# 步骤 1：每个服务器的 current_weight += weight
for s in servers:
s.current_weight += s.weight
before = [f"{s.name}={s.current_weight}" for s in servers]
# 步骤 2：选择 current_weight 最大的
selected = max(servers, key=lambda s: s.current_weight)
# 步骤 3：被选中的 current_weight -= total_weight
selected.current_weight -= total_weight
after = [f"{s.name}={s.current_weight}" for s in servers]
print(f" {i+1:>2} | {str(before):>30} | {selected.name:>6} | {str(after):>30}")
results.append(selected.name)
return results
# 模拟 Nginx 的加权轮询
servers = [
Server("A", 5),
Server("B", 3),
Server("C", 2),
]
results = smooth_weighted_round_robin(servers, 10)
print(f"\n请求分配序列: {results}")
print(f"A 收到: {results.count('A')} 次 (权重 5, 期望 50%)")
print(f"B 收到: {results.count('B')} 次 (权重 3, 期望 30%)")
print(f"C 收到: {results.count('C')} 次 (权重 2, 期望 20%)")
输出：
服务器: [('A', '权重=5'), ('B', '权重=3'), ('C', '权重=2')]
总权重: 10
请求 | 选中前 current_weight | 选中 | 选中后 current_weight
------------------------------------------------------------------------------------------
1 | ['A=5', 'B=3', 'C=2'] | A | ['A=-5', 'B=3', 'C=2']
2 | ['A=0', 'B=6', 'C=4'] | B | ['A=0', 'B=-4', 'C=4']
3 | ['A=5', 'B=-1', 'C=6'] | C | ['A=5', 'B=-1', 'C=-4']
4 | ['A=10', 'B=2', 'C=-2'] | A | ['A=0', 'B=2', 'C=-2']
5 | ['A=5', 'B=5', 'C=0'] | A | ['A=-5', 'B=5', 'C=0']
6 | ['A=0', 'B=8', 'C=2'] | B | ['A=0', 'B=-2', 'C=2']
7 | ['A=5', 'B=1', 'C=4'] | A | ['A=-5', 'B=1', 'C=4']
8 | ['A=0', 'B=4', 'C=6'] | C | ['A=0', 'B=4', 'C=-4']
9 | ['A=5', 'B=7', 'C=-2'] | B | ['A=5', 'B=-3', 'C=-2']
10 | ['A=10', 'B=0', 'C=0'] | A | ['A=0', 'B=0', 'C=0']
请求分配序列: ['A', 'B', 'C', 'A', 'A', 'B', 'A', 'C', 'B', 'A']
A 收到: 5 次 (权重 5, 期望 50%) ✅
B 收到: 3 次 (权重 3, 期望 30%) ✅
C 收到: 2 次 (权重 2, 期望 20%) ✅
注意第 10 次后，所有
current_weight归零，进入下一个循环。分布非常平滑——A 不会连续出现 5 次！
3. IP Hash 算法实现
import hashlib
class IPHashBalancer:
"""Nginx 的 IP Hash 负载均衡"""
def __init__(self, servers):
self.servers = servers
def select(self, client_ip):
"""同一个 IP 总是选到同一个后端"""
# Nginx 实际用的是 IP 的前三段做 hash
# 比如 192.168.1.100 和 192.168.1.200 会 hash 到同一个值
ip_parts = client_ip.split(".")
ip_key = ".".join(ip_parts[:3]) # 取前三段
hash_value = 0
for char in ip_key:
hash_value = (hash_value * 113 + ord(char)) & 0x7FFFFFFF
index = hash_value % len(self.servers)
return self.servers[index]
# 测试
servers = ["Server-A", "Server-B", "Server-C"]
balancer = IPHashBalancer(servers)
test_ips = [
"192.168.1.100",
"192.168.1.200", # 同一网段，会分到同一个服务器
"10.0.0.50",
"172.16.5.10",
"192.168.1.100", # 重复 IP，结果不变
]
print("IP Hash 负载均衡:")
for ip in test_ips:
server = balancer.select(ip)
print(f" {ip:>20} → {server}")
输出：
IP Hash 负载均衡:
192.168.1.100 → Server-B
192.168.1.200 → Server-B ← 同网段，同服务器
10.0.0.50 → Server-A
172.16.5.10 → Server-C
192.168.1.100 → Server-B ← 同 IP，结果不变！（会话保持）
4. 最少连接算法实现
import random
import threading
class LeastConnBalancer:
"""Nginx 的最少连接负载均衡"""
def __init__(self, servers_config):
"""servers_config: [(name, weight), ...]"""
self.servers = []
for name, weight in servers_config:
self.servers.append({
"name": name,
"weight": weight,
"active_connections": 0,
})
def select(self):
"""选择 active_connections / weight 最小的服务器"""
best = None
best_score = float("inf")
for server in self.servers:
# 核心公式：连接数 / 权重 → 值越小越优先
score = server["active_connections"] / server["weight"]
if score < best_score:
best_score = score
best = server
best["active_connections"] += 1
return best
def release(self, server):
"""请求完成，释放连接"""
server["active_connections"] -= 1
# 模拟
balancer = LeastConnBalancer([
("A-高配", 5),
("B-中配", 3),
("C-低配", 1),
])
print("最少连接负载均衡模拟:\n")
# 模拟 20 个请求
active_requests = []
for i in range(20):
server = balancer.select()
active_requests.append(server)
status = " | ".join(
f"{s['name']}: {s['active_connections']}连接"
for s in balancer.servers
)
print(f" 请求{i+1:>2} → {server['name']:>6} [{status}]")
# 随机释放一些老请求（模拟请求完成）
if len(active_requests) > 5 and random.random() > 0.5:
released = active_requests.pop(0)
balancer.release(released)
5. 一致性 Hash 算法（第三方模块）
import hashlib
import bisect
class ConsistentHashBalancer:
"""
一致性 Hash 负载均衡
用于需要缓存亲和性的场景：
同一个 URL 总是路由到同一个后端 → 提高缓存命中率
"""
def __init__(self, servers, replicas=150):
self.replicas = replicas # 每个服务器的虚拟节点数
self.ring = [] # Hash 环
self.server_map = {} # hash值 → 服务器名
for server in servers:
self._add_server(server)
self.ring.sort()
def _hash(self, key):
return int(hashlib.md5(key.encode()).hexdigest(), 16)
def _add_server(self, server):
"""在 Hash 环上添加虚拟节点"""
for i in range(self.replicas):
virtual_key = f"{server}#{i}"
h = self._hash(virtual_key)
self.ring.append(h)
self.server_map[h] = server
def select(self, key):
"""根据 key 找到顺时针最近的服务器"""
if not self.ring:
return None
h = self._hash(key)
# 在有序环上二分查找
idx = bisect.bisect_left(self.ring, h)
if idx == len(self.ring):
idx = 0 # 绕回环的起点
return self.server_map[self.ring[idx]]
# 测试
servers = ["Server-A", "Server-B", "Server-C"]
balancer = ConsistentHashBalancer(servers)
# 同一个 URL 总是路由到同一个后端
urls = ["/api/user/1", "/api/user/2", "/api/order/100", "/api/user/1", "/static/logo.png"]
print("一致性 Hash（相同 URL → 相同后端 → 高缓存命中率）:")
for url in urls:
server = balancer.select(url)
print(f" {url:>20} → {server}")
# 模拟移除一台服务器后的影响
print(f"\n假设 Server-C 宕机，重新分配:")
balancer2 = ConsistentHashBalancer(["Server-A", "Server-B"])
changed = 0
for url in [f"/page/{i}" for i in range(100)]:
old = balancer.select(url)
new = balancer2.select(url)
if old != new:
changed += 1
print(f" 100 个 URL 中只有 {changed} 个被重新分配（约 {changed}%）")
print(f" 传统 Hash 取模：移除一台后几乎所有请求都要重新分配！")
输出：
一致性 Hash（相同 URL → 相同后端 → 高缓存命中率）:
/api/user/1 → Server-B
/api/user/2 → Server-A
/api/order/100 → Server-C
/api/user/1 → Server-B ← 同 URL 同后端
/static/logo.png → Server-A
假设 Server-C 宕机，重新分配:
100 个 URL 中只有 33 个被重新分配（约 33%）
传统 Hash 取模：移除一台后几乎所有请求都要重新分配！
6. 健康检查与故障转移
import time
import random
class HealthCheckBalancer:
"""模拟 Nginx 的被动健康检查"""
def __init__(self, servers_config):
self.servers = []
for name, weight in servers_config:
self.servers.append({
"name": name,
"weight": weight,
"fails": 0, # 连续失败次数
"max_fails": 3, # 最大失败次数（超过后标记为 down）
"down": False, # 是否下线
"down_since": None, # 下线时间
"fail_timeout": 10, # 下线恢复间隔（秒）
})
def select(self):
"""选择一个可用的后端"""
now = time.time()
available = []
for s in self.servers:
if s["down"]:
# 检查是否到了重试时间
if now - s["down_since"] >= s["fail_timeout"]:
print(f" 🔄 {s['name']} 下线已超过 {s['fail_timeout']}s，尝试恢复")
s["down"] = False
s["fails"] = 0
available.append(s)
else:
available.append(s)
if not available:
print(" ⚠️ 所有服务器都不可用！使用备用服务器")
return None
# 简单轮询选择
return available[0] # 实际会用加权轮询
def report_success(self, server):
"""请求成功"""
server["fails"] = 0
def report_failure(self, server):
"""请求失败"""
server["fails"] += 1
if server["fails"] >= server["max_fails"]:
server["down"] = True
server["down_since"] = time.time()
print(f" ❌ {server['name']} 连续失败 {server['fails']} 次，标记为下线！")
# 模拟
balancer = HealthCheckBalancer([
("Server-A", 5),
("Server-B", 3),
])
print("被动健康检查模拟:\n")
for i in range(8):
server = balancer.select()
if server is None:
print(f" 请求 {i+1}: 无可用服务器")
continue
# 模拟 Server-A 连续失败
if server["name"] == "Server-A" and i < 5:
success = False
else:
success = True
if success:
balancer.report_success(server)
print(f" 请求 {i+1}: {server['name']} ✅ 成功")
else:
balancer.report_failure(server)
print(f" 请求 {i+1}: {server['name']} ❌ 失败 (连续失败 {server['fails']} 次)")
对应的 Nginx 配置：
upstream backend {
server 192.168.1.101:8000 max_fails=3 fail_timeout=10s;
server 192.168.1.102:8000 max_fails=3 fail_timeout=10s;
server 192.168.1.103:8000 backup; # 备用服务器，只在上面都挂了才用
}
三、完整模拟：Nginx 反向代理 + 负载均衡
import time
import random
from collections import defaultdict
class NginxSimulator:
"""模拟 Nginx 反向代理 + 负载均衡的完整流程"""
def __init__(self, upstream_config):
self.servers = []
for name, weight in upstream_config:
self.servers.append({
"name": name,
"weight": weight,
"current_weight": 0,
"active_connections": 0,
"total_requests": 0,
"fails": 0,
"max_fails": 3,
"down": False,
})
self.total_weight = sum(s["weight"] for s in self.servers)
self.stats = defaultdict(int)
def _select_server_wrr(self):
"""加权轮询选择后端"""
available = [s for s in self.servers if not s["down"]]
if not available:
return None
total = sum(s["weight"] for s in available)
for s in available:
s["current_weight"] += s["weight"]
selected = max(available, key=lambda s: s["current_weight"])
selected["current_weight"] -= total
return selected
def handle_request(self, method, uri, client_ip, headers=None):
"""
模拟一次完整的反向代理请求
"""
headers = headers or {}
# ===== 阶段 1：接收客户端请求 =====
print(f"\n{'='*60}")
print(f"📥 收到请求: {method} {uri}")
print(f" 客户端 IP: {client_ip}")
# ===== 阶段 2：匹配 location =====
backend_prefix = self._match_location(uri)
print(f" 匹配 location: {backend_prefix}")
# ===== 阶段 3：负载均衡选择后端 =====
server = self._select_server_wrr()
if not server:
print(f" ❌ 502 Bad Gateway - 无可用后端")
return 502
print(f" 选择后端: {server['name']}")
server["active_connections"] += 1
server["total_requests"] += 1
# ===== 阶段 4：构造代理请求 =====
proxy_headers = {
"Host": headers.get("Host", "backend"),
"X-Real-IP": client_ip,
"X-Forwarded-For": client_ip,
"X-Forwarded-Proto": "https",
"Connection": "", # 清除以支持 keepalive
}
print(f" 修改请求头: X-Real-IP={client_ip}")
print(f" 转发到: http://{server['name']}{uri}")
# ===== 阶段 5：模拟后端响应 =====
response_time = random.uniform(0.01, 0.1)
# 模拟偶尔失败
is_success = random.random() > 0.1 # 10% 失败率
if is_success:
server["fails"] = 0
status = 200
print(f" ✅ 后端响应: {status} ({response_time*1000:.0f}ms)")
else:
server["fails"] += 1
status = 502
print(f" ❌ 后端失败: 502 (连续失败 {server['fails']}/{server['max_fails']})")
if server["fails"] >= server["max_fails"]:
server["down"] = True
print(f" ⛔ {server['name']} 被标记为下线!")
# 尝试下一个服务器（failover）
next_server = self._select_server_wrr()
if next_server:
print(f" 🔄 故障转移到: {next_server['name']}")
status = 200
server["active_connections"] -= 1
# ===== 阶段 6：返回响应给客户端 =====
self.stats[status] += 1
print(f" 📤 返回客户端: {status}")
return status
def _match_location(self, uri):
"""模拟 location 匹配"""
if uri.startswith("/api/"):
return "proxy_pass → backend"
elif uri.startswith("/static/"):
return "root → /var/www/static"
else:
return "proxy_pass → backend (default)"
def print_stats(self):
"""打印统计信息"""
print(f"\n{'='*60}")
print("📊 统计信息:\n")
for s in self.servers:
bar = "█" * s["total_requests"]
status = "🔴 下线" if s["down"] else "🟢 正常"
print(f" {s['name']:>10} | 权重={s['weight']} | "
f"请求={s['total_requests']:>3} | {status} | {bar}")
print(f"\n 响应状态码: {dict(self.stats)}")
# ===== 运行模拟 =====
if __name__ == "__main__":
nginx = NginxSimulator([
("Backend-A", 5),
("Backend-B", 3),
("Backend-C", 2),
])
# 模拟 15 个请求
uris = [
"/api/users", "/api/orders", "/api/products",
"/static/logo.png", "/api/login", "/api/profile",
"/api/settings", "/api/users/1", "/api/orders/5",
"/", "/api/health", "/api/search",
"/api/upload", "/api/notify", "/api/logout",
]
for uri in uris:
ip = f"192.168.1.{random.randint(1, 255)}"
nginx.handle_request("GET", uri, ip, {"Host": "www.example.com"})
time.sleep(0.05)
nginx.print_stats()
四、算法选择指南
你的需求是什么？
│
├── 后端性能相同？
│ └── ✅ 轮询 (round-robin)
│
├── 后端性能不同？
│ └── ✅ 加权轮询 (weighted round-robin)
│
├── 需要会话保持？（用户登录状态）
│ └── ✅ IP Hash
│
├── 请求处理时间差异很大？
│ └── ✅ 最少连接 (least_conn)
│
├── 有缓存服务器，要提高命中率？
│ └── ✅ 一致性 Hash (需第三方模块)
│
└── 不确定？
└── ✅ 加权轮询（Nginx 默认，适用最广）
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
8
0- 0
已为社区贡献6条内容
所有评论(0)