负载均衡

最少连接均衡根据活动连接数选择节点，并遵循具体实现的权重和平局规则。在 nginx 中，least_conn 考虑服务器权重，对于同样合适的节点使用加权轮询。这有助于处理占用连接时间不同的请求，但仍然只是连接维度的测量，不直接知道 CPU、内存压力、排队任务或数据库竞争。算法是否合适，取决于该信号与实际瓶颈有多大关联。 坏掉的样子 把选中的节点称为“最空的服务器”，夸大了信号含义。一条连接可能承载昂贵操作，另一条只是便宜的等待。HTTP 响应关闭后，后台工作还可能继续；多路复用协议也可能把多个操作放在一条连接上。因此，连接少并不能证明后端有足够余量接收下一个昂贵请求。需要保护哪种资源，必须先明确，而不能让算法名称替代容量分析。 计数还是均衡器观察到的上游状态，不是对后端全部客户端的全知统计。通过另一个代理或内部直连到来的工作，可能不在当前视野中。在 nginx 中，worker 范围和上游共享区域配置影响哪些运行状态共享；独立均衡器也不会自动合并观察，形成一个全局最少连接决定。忽略计数范围，会让局部正确的选择被误解成整个集群的最优分配。 权重使只比较原始计数不够准确。权重二的节点可以合理承载比权重一节点更多的活动连接。修改权重表达一种容量假设，并没有测量该假设是否真实。空闲 keepalive 缓存又是另一种数量，不应与这里参与选择的活动连接混淆。应分别识别正在占用的交换、为复用保留的空闲连接，以及真正消耗计算资源的工作。 快速验证 将脚本保存为 check-least-connections.py，准备 Python、Docker 和固定镜像后执行。记录环境使用单个 nginx 1.28.0 worker，没有配置上游 keepalive 缓存，并连接不同端口上的两个 HTTP 节点。后端提供受控响应开关与活动计数，让保持中的工作可以直接观察，不依赖猜测某个睡眠时间是否足够长： from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,json,socket,subprocess,time,uuid NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def wait(predicate): deadline=time.monotonic()+15 while time.monotonic()<deadline: try: if predicate():return except OSError:pass time.sleep(.1) raise AssertionError('fixture did not become ready') @contextmanager def lab(configuration,files=None,backend=None,extra_ports=()): name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend'] try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration) for path,content in (files or {}).items(): target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True) target.write_bytes(content if isinstance(content,bytes) else content.encode()) docker('network','create',name) if backend is not None: (root/'backend.py').write_text(backend) (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n') docker('build','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root)) docker('run','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1]) wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--network','none','-t',tags[0],str(root)) docker('run','--rm','--network',name,tags[0],'nginx','-t') published=[arg for value in (8080,*extra_ports) for arg in ('-p','127.0.0.1::'+str(value))] docker('run','-d','--name',containers[0],'--network',name,*published,tags[0]) details=json.loads(docker('inspect',containers[0]).stdout)[0] binding=details['NetworkSettings']['Ports']['8080/tcp'][0];assert binding['HostIp']=='127.0.0.1';port=int(binding['HostPort']) def ready(): connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True wait(ready) def request(path,method='GET',headers=None,body=None): connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5) try: connection.request(method,path,body=body,headers=headers or {}) response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read() finally:connection.close() request.port=port;request.network=name request.ports={int(key.split('/')[0]):int(value[0]['HostPort']) for key,value in details['NetworkSettings']['Ports'].items() if value} yield containers[0],request finally: docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False) def config(locations,extra=''): return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n' backend=r''' from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer from urllib.parse import urlsplit,parse_qs import itertools,json,threading lock=threading.Lock();ids=itertools.count(1);counts={'A':0,'B':0};active={'A':0,'B':0};faults={'A':200,'B':200};gates={} class Handler(BaseHTTPRequestHandler): protocol_version='HTTP/1.1' def setup(self):super().setup();self.ident=next(ids) def log_message(self,*args):pass def do_GET(self): parsed=urlsplit(self.path);label=self.server.label;code=200 if label=='control': query=parse_qs(parsed.query) with lock: if parsed.path=='/fail':faults[query['peer'][0]]=int(query['status'][0]) if parsed.path=='/release':gates[query['path'][0]].set() data={'counts':dict(counts),'active':dict(active),'faults':dict(faults),'held':list(gates)} else: with lock: counts[label]+=1;active[label]+=1;code=faults[label] if parsed.path.startswith('/hold/'):gates[parsed.path]=threading.Event() if parsed.path.startswith('/hold/'):assert gates[parsed.path].wait(20) data={'peer':label,'path':self.path,'connection':self.ident} body=json.dumps(data,sort_keys=True).encode();self.send_response(code) self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(body))) self.end_headers();self.wfile.write(body) if label!='control': with lock:active[label]-=1 for port,label in [(8081,'A'),(8082,'B'),(8083,'control')]: server=ThreadingHTTPServer(('0.0.0.0',port),Handler);server.label=label threading.Thread(target=server.serve_forever,daemon=True).start() threading.Event().wait() ''' from concurrent.futures import ThreadPoolExecutor extra=''' upstream least { least_conn; server backend:8081; server backend:8082; } upstream weighted_least { least_conn; server backend:8081 weight=2; server backend:8082; } ''' locations=''' location /least/ { proxy_pass http://least/; } location /weighted/ { proxy_pass http://weighted_least/; } location /control/ { proxy_pass http://backend:8083/; } ''' with lab(config(locations,extra),backend=backend) as (_,request): def get(path): status,_,body=request(path);assert status==200;return json.loads(body) def state():return get('/control/state') results={} with ThreadPoolExecutor(max_workers=3) as pool: held=pool.submit(get,'/least/hold/first');wait(lambda:sum(state()['active'].values())==1) occupied=next(peer for peer,value in state()['active'].items() if value==1) quick=[get('/least/quick')['peer'] for _ in range(6)] assert all(peer!=occupied for peer in quick),quick get('/control/release?path=/hold/first');assert held.result()['peer']==occupied wait(lambda:sum(state()['active'].values())==0) results['held_peer']=occupied;results['quick_while_held']=quick futures=[] for index in range(3): futures.append(pool.submit(get,'/weighted/hold/w'+str(index))) wait(lambda:sum(state()['active'].values())==index+1) snapshot=state()['active'];assert snapshot=={'A':2,'B':1},snapshot for index in range(3):get('/control/release?path=/hold/w'+str(index)) results['weighted_active']=snapshot;results['weighted_completed']=[f.result()['peer'] for f in futures] wait(lambda:sum(state()['active'].values())==0) print(json.dumps(results,sort_keys=True)) 第一个请求进入同权最少连接组，并在某个后端处理器中等待。控制端先观察恰好一个活动请求，并识别被占用节点，然后才发送其他请求。初始平局究竟由哪台节点胜出不是主要断言；实验从后端真实状态发现该身份，不把某个偶然的初始顺序写成算法必须遵守的规则。 ...

upstream 组列出候选后端端点，均衡方法从符合条件的节点中选择目标。轮询按权重分配选择，最少连接考虑活动连接与权重，哈希方法则根据指定 key 建立亲和性。这些决定描述路由策略，不会自动测量 CPU 利用率，不保证响应时间相等，也不会在节点消失时让本地会话状态自动获得持久性。选择方法之前，应先明确希望它依据什么信号决策。 坏掉的样子 常见测量错误是看错地址。curl 的 remote IP 标识 curl 实际连接的服务器，在这种架构里通常就是代理。访问日志里的客户端地址字段，同样不是后端分布报告。要评估节点选择，应查看真正的上游地址或受控后端身份；如果重试可能访问多个节点，还要保留各次尝试信息。不能把入口连接信息当成后端选择结果。 请求数均匀不代表工作量均匀。长报表和很小的缓存响应都算一个请求，却可能占用完全不同的资源。配置权重表达预期相对份额，不是自动测得的机器容量。把权重翻倍，不能证明不同后端的每核负载或延迟已经相等。权重是否合理，需要代表性请求成本和节点能力支持，而不是只依据某个硬件参数作推断。 亲和性有另一组失败方式。很多用户可能共享同一源地址，所以 IP 哈希可能集中流量，而不是按个人均匀分布。节点失败或成员变化会改变落点，本地会话状态不会仅因为请求换了机器就一起移动。ip_hash 也不应被描述成 nginx 可配置的一致性哈希环；显式 hash ... consistent 是另一种独立方法，不能把两种策略的性质混用。 快速验证 将脚本保存为 check-balancing.py，准备 Python、Docker 和固定镜像后执行。记录环境使用单个 nginx 1.28.0 worker，以及一个可丢弃后端容器中不同端口上的两个可识别 HTTP 节点。共享机器是有意安排：实验测量选择结果，不比较硬件容量。客户端只通过宿主回环访问，结束后清理自己的资源： from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,json,socket,subprocess,time,uuid NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def wait(predicate): deadline=time.monotonic()+15 while time.monotonic()<deadline: try: if predicate():return except OSError:pass time.sleep(.1) raise AssertionError('fixture did not become ready') @contextmanager def lab(configuration,files=None,backend=None,extra_ports=()): name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend'] try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration) for path,content in (files or {}).items(): target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True) target.write_bytes(content if isinstance(content,bytes) else content.encode()) docker('network','create',name) if backend is not None: (root/'backend.py').write_text(backend) (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n') docker('build','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root)) docker('run','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1]) wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--network','none','-t',tags[0],str(root)) docker('run','--rm','--network',name,tags[0],'nginx','-t') published=[arg for value in (8080,*extra_ports) for arg in ('-p','127.0.0.1::'+str(value))] docker('run','-d','--name',containers[0],'--network',name,*published,tags[0]) details=json.loads(docker('inspect',containers[0]).stdout)[0] binding=details['NetworkSettings']['Ports']['8080/tcp'][0];assert binding['HostIp']=='127.0.0.1';port=int(binding['HostPort']) def ready(): connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True wait(ready) def request(path,method='GET',headers=None,body=None): connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5) try: connection.request(method,path,body=body,headers=headers or {}) response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read() finally:connection.close() request.port=port;request.network=name request.ports={int(key.split('/')[0]):int(value[0]['HostPort']) for key,value in details['NetworkSettings']['Ports'].items() if value} yield containers[0],request finally: docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False) def config(locations,extra=''): return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n' backend=r''' from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer from urllib.parse import urlsplit,parse_qs import itertools,json,threading lock=threading.Lock();ids=itertools.count(1);counts={'A':0,'B':0};active={'A':0,'B':0};faults={'A':200,'B':200};gates={} class Handler(BaseHTTPRequestHandler): protocol_version='HTTP/1.1' def setup(self):super().setup();self.ident=next(ids) def log_message(self,*args):pass def do_GET(self): parsed=urlsplit(self.path);label=self.server.label;code=200 if label=='control': query=parse_qs(parsed.query) with lock: if parsed.path=='/fail':faults[query['peer'][0]]=int(query['status'][0]) if parsed.path=='/release':gates[query['path'][0]].set() data={'counts':dict(counts),'active':dict(active),'faults':dict(faults),'held':list(gates)} else: with lock: counts[label]+=1;active[label]+=1;code=faults[label] if parsed.path.startswith('/hold/'):gates[parsed.path]=threading.Event() if parsed.path.startswith('/hold/'):assert gates[parsed.path].wait(20) data={'peer':label,'path':self.path,'connection':self.ident} body=json.dumps(data,sort_keys=True).encode();self.send_response(code) self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(body))) self.end_headers();self.wfile.write(body) if label!='control': with lock:active[label]-=1 for port,label in [(8081,'A'),(8082,'B'),(8083,'control')]: server=ThreadingHTTPServer(('0.0.0.0',port),Handler);server.label=label threading.Thread(target=server.serve_forever,daemon=True).start() threading.Event().wait() ''' from collections import Counter extra=''' upstream round { server backend:8081; server backend:8082; } upstream weighted { server backend:8081 weight=2; server backend:8082; } upstream by_ip { ip_hash; server backend:8081; server backend:8082; } upstream by_key { hash $arg_key consistent; server backend:8081; server backend:8082; } add_header X-Upstream $upstream_addr always; add_header X-Peer $remote_addr always; ''' locations='\n'.join(f'location /{path}/ {{ proxy_pass http://{group}/; }}' for path,group in [('round','round'),('weighted','weighted'),('sticky','by_ip'),('key','by_key')]) with lab(config(locations,extra),backend=backend) as (_,request): def get(path,headers=None): status,h,b=request(path,headers=headers);assert status==200;data=json.loads(b) assert h['x-upstream'].endswith(':8081' if data['peer']=='A' else ':8082') return data['peer'],h results={} for mode,expected in [('round',{'A':6,'B':6}),('weighted',{'A':8,'B':4})]: chosen=[get('/'+mode+'/item')[0] for _ in range(12)] assert dict(Counter(chosen))==expected,chosen;results[mode]={'sequence':chosen,'counts':dict(Counter(chosen))} sticky=[get('/sticky/item',{'X-Forwarded-For':'192.0.2.'+str(i)}) for i in range(1,9)] assert len({r[0] for r in sticky})==len({r[1]['x-peer'] for r in sticky})==1 results['ip_hash']={'peers':[r[0] for r in sticky],'socket_peer':sticky[0][1]['x-peer']} groups={} for key in ['alpha','beta','gamma','delta','epsilon','zeta']: pair=[get('/key/item?key='+key)[0] for _ in range(2)];assert pair[0]==pair[1];groups[key]=pair[0] assert set(groups.values())=={'A','B'},groups;results['consistent_key_hash']=groups print(json.dumps(results,sort_keys=True)) 对同权轮询组顺序发送十二次请求，A、B 各返回六次，本次受控运行中两者交替出现。每个响应的后端标记都会与 nginx 上游地址头中的端口对应检查，因此观察的是实际选中节点，而不是从客户端连接 nginx 的地址推测分布。即使所有请求都经过同一入口，也能明确区分最终由哪个后端回答。 ...

nginx 被动上游健康判断利用真实请求结果，决定某个节点是否暂时停止接收流量。max_fails 设置失败尝试阈值，fail_timeout 同时规定计数区间和不可用时长。这些配置不会安排周期性的应用探测。主动检查需要被支持的检查能力与额外配置，不能把真实请求结果策略直接称作一个已经运行的探活定时器。 坏掉的样子 不是每个看起来像错误的 HTTP 响应，都算一次失败上游尝试。代理重试配置参与定义这种分类。在本次设置中，只有列入 http_503，HTTP 503 才参与相应被动失败处理。如果认为所有 5xx 都会自动导致健康失败，就可能期待节点已被摘除，而 nginx 仍依据真实规则继续选择它。必须先核对什么结果被配置认定为失败。 另一个错误是把超时窗口结束当成恢复证明。不可用时段结束后，节点可以重新获得真实流量资格，未必先通过独立健康探测。如果后端仍坏着，这次请求仍会失败。least_conn 依据活动连接改变选择，并不提供渐进恢复权重的升温过程，也不能代替受支持的 slow-start 机制。资格恢复、健康证明和逐渐加量需要分别理解。 重试也有边界。响应数据已经发送给客户端后，更换节点不能透明修复中途错误；重放有副作用的操作还可能产生重复。下面只使用 GET，并在成功正文发送之前产生失败，且明确限制尝试次数。它没有为所有应用方法证明安全重试语义，不能把这个成功切换案例直接用于任何写操作。 快速验证 将脚本保存为 check-passive-health.py，准备 Python、Docker 和固定镜像后运行。记录服务器为单 worker 的 nginx 1.28.0。可丢弃容器里的两个 HTTP 端点报告自身身份，并能受控返回 503。另一个控制监听提供计数，查询它不会通过被均衡组产生业务请求，便于区分观察动作与真实探测流量： from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,json,socket,subprocess,time,uuid NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def wait(predicate): deadline=time.monotonic()+15 while time.monotonic()<deadline: try: if predicate():return except OSError:pass time.sleep(.1) raise AssertionError('fixture did not become ready') @contextmanager def lab(configuration,files=None,backend=None,extra_ports=()): name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend'] try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration) for path,content in (files or {}).items(): target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True) target.write_bytes(content if isinstance(content,bytes) else content.encode()) docker('network','create',name) if backend is not None: (root/'backend.py').write_text(backend) (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n') docker('build','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root)) docker('run','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1]) wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--network','none','-t',tags[0],str(root)) docker('run','--rm','--network',name,tags[0],'nginx','-t') published=[arg for value in (8080,*extra_ports) for arg in ('-p','127.0.0.1::'+str(value))] docker('run','-d','--name',containers[0],'--network',name,*published,tags[0]) details=json.loads(docker('inspect',containers[0]).stdout)[0] binding=details['NetworkSettings']['Ports']['8080/tcp'][0];assert binding['HostIp']=='127.0.0.1';port=int(binding['HostPort']) def ready(): connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True wait(ready) def request(path,method='GET',headers=None,body=None): connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5) try: connection.request(method,path,body=body,headers=headers or {}) response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read() finally:connection.close() request.port=port;request.network=name request.ports={int(key.split('/')[0]):int(value[0]['HostPort']) for key,value in details['NetworkSettings']['Ports'].items() if value} yield containers[0],request finally: docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False) def config(locations,extra=''): return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n' backend=r''' from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer from urllib.parse import urlsplit,parse_qs import itertools,json,threading lock=threading.Lock();ids=itertools.count(1);counts={'A':0,'B':0};active={'A':0,'B':0};faults={'A':200,'B':200};gates={} class Handler(BaseHTTPRequestHandler): protocol_version='HTTP/1.1' def setup(self):super().setup();self.ident=next(ids) def log_message(self,*args):pass def do_GET(self): parsed=urlsplit(self.path);label=self.server.label;code=200 if label=='control': query=parse_qs(parsed.query) with lock: if parsed.path=='/fail':faults[query['peer'][0]]=int(query['status'][0]) if parsed.path=='/release':gates[query['path'][0]].set() data={'counts':dict(counts),'active':dict(active),'faults':dict(faults),'held':list(gates)} else: with lock: counts[label]+=1;active[label]+=1;code=faults[label] if parsed.path.startswith('/hold/'):gates[parsed.path]=threading.Event() if parsed.path.startswith('/hold/'):assert gates[parsed.path].wait(20) data={'peer':label,'path':self.path,'connection':self.ident} body=json.dumps(data,sort_keys=True).encode();self.send_response(code) self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(body))) self.end_headers();self.wfile.write(body) if label!='control': with lock:active[label]-=1 for port,label in [(8081,'A'),(8082,'B'),(8083,'control')]: server=ThreadingHTTPServer(('0.0.0.0',port),Handler);server.label=label threading.Thread(target=server.serve_forever,daemon=True).start() threading.Event().wait() ''' extra=''' upstream passive { server backend:8081 max_fails=1 fail_timeout=2s; server backend:8082; } upstream plain { server backend:8081 max_fails=1 fail_timeout=30s; server backend:8082; } upstream single { server backend:8081 max_fails=1 fail_timeout=30s; } add_header X-Upstream $upstream_addr always; add_header X-Attempts $upstream_status always; ''' locations=''' location /passive { proxy_next_upstream error timeout http_503; proxy_next_upstream_tries 2; proxy_pass http://passive; } location /plain { proxy_next_upstream error timeout; proxy_pass http://plain; } location /single { proxy_next_upstream error timeout http_503; proxy_pass http://single; } location /control/ { proxy_pass http://backend:8083/; } ''' active_config=config('location / { proxy_pass http://peers; health_check; }', 'upstream peers { zone peers 64k; server backend:8081; server backend:8082; }') with lab(config(locations,extra),{'active.conf':active_config},backend) as (proxy,request): def control(path): status,_,body=request('/control/'+path);assert status==200;return json.loads(body) def call(path): status,h,b=request(path);return {'status':status,'attempts':h['x-attempts'],'upstream':h['x-upstream'],'peer':json.loads(b)['peer']} control('fail?peer=A&status=503');results={} plain=[call('/plain') for _ in range(6)] assert [r['status'] for r in plain].count(503)==3 and all(',' not in r['attempts'] for r in plain) results['unlisted_http_503']=plain for _ in range(2): first=call('/passive') if first['attempts']=='503, 200':break assert first['status']==200 and first['attempts']=='503, 200' and first['peer']=='B',first results['failed_attempt_then_success']=first after_failure=control('state')['counts']['A'] excluded=[call('/passive') for _ in range(4)] assert all(r['peer']=='B' and r['attempts']=='200' for r in excluded) assert control('state')['counts']['A']==after_failure time.sleep(.4);assert control('state')['counts']['A']==after_failure results['during_exclusion']=excluded;results['background_probe_count']=0 control('fail?peer=A&status=200');deadline=time.monotonic()+6;recovered=None while time.monotonic()<deadline: observed=call('/passive') if observed['peer']=='A':recovered=observed;break time.sleep(.1) assert recovered and recovered['status']==200;results['live_request_after_window']=recovered control('fail?peer=A&status=503');before=control('state')['counts']['A'] single=[call('/single') for _ in range(2)] assert all(r['status']==503 and r['attempts']=='503' for r in single) assert control('state')['counts']['A']==before+2;results['single_peer_still_attempted']=single probe=docker('exec',proxy,'nginx','-t','-c','/srv/active.conf',check=False) assert probe.returncode!=0 and 'unknown directive "health_check"' in probe.stderr results['active_health_check_directive_supported']=False print(json.dumps(results,sort_keys=True)) 首先让 A 返回真实 HTTP 503，B 返回 200。一个对照组只对连接错误和超时重试。六次请求仍各选择 A、B 三次，没有多次尝试的响应。A 的 max_fails=1 不会因为这个未列入的 HTTP 结果把它移除。实验检查状态与所选上游地址，不只查看最终正文，从而避免把应用返回值与节点资格变化混淆。 ...

传输层均衡与 HTTP 均衡的实用区别，在于什么位置可以作出路由决定。TCP 转发可以为一条连接选择后端，然后不解释 HTTP 请求，直接转发字节流。HTTP 代理则解析请求，并能根据路径或请求头选择上游。因此，同一个客户端连接里的多个请求，可能获得不同路由结果。选择策略前，应先明确需要按连接还是按请求决定目标。 坏掉的样子 如果期待 TCP 连接选择还能分别路由内部每个 URL，请求就可能到达错误应用。字节流已经转发到一个后端后，后续 HTTP 路径不会自动触发重新选节点。反过来，如果把 HTTP 路由当成永久连接亲和，也可能让应用误以为同一客户端连接的每个请求都会到达同一份本地状态。两种假设都需要通过实际连接复用案例验证。 层级名称也不能完整说明 TLS 行为。TLS 终止与 HTTP 解释是不同能力。stream 代理可以终止 TLS，却不成为 HTTP 路径路由器；也可以不解密应用数据，只查看部分 ClientHello 元数据。按 SNI 选择目标，并不等于能看到加密 URL、请求头或响应正文。应描述实际配置行为，而不是认为产品标签已经证明所有这些性质。 性能结论同样需要谨慎。用户态 TCP 转发与内核包转发具有不同资源路径，即使两者都被称为传输层均衡。HTTP 代理会执行解析和策略工作，但相对吞吐取决于实现与负载。仅选择 L4 或 L7 这两个术语，不能推出通用的线速能力或固定额外开销；需要以真实路径和代表性流量测量，而不是用层级名称代替基准。 快速验证 将脚本保存为 check-routing-layers.py，准备 Python、Docker 和固定镜像后执行。记录服务器为具有 stream 模块的 nginx 1.28.0。两个 HTTP 节点暴露不同身份和后端连接编号。nginx 分别提供 HTTP 路由监听与 TCP 转发监听，都只发布到宿主回环，使相同请求序列可以通过两条路径比较： ...