最少连接均衡根据活动连接数选择节点，并遵循具体实现的权重和平局规则。在 nginx 中，least_conn 考虑服务器权重，对于同样合适的节点使用加权轮询。这有助于处理占用连接时间不同的请求，但仍然只是连接维度的测量，不直接知道 CPU、内存压力、排队任务或数据库竞争。算法是否合适，取决于该信号与实际瓶颈有多大关联。

坏掉的样子

把选中的节点称为“最空的服务器”，夸大了信号含义。一条连接可能承载昂贵操作，另一条只是便宜的等待。HTTP 响应关闭后，后台工作还可能继续；多路复用协议也可能把多个操作放在一条连接上。因此，连接少并不能证明后端有足够余量接收下一个昂贵请求。需要保护哪种资源，必须先明确，而不能让算法名称替代容量分析。

计数还是均衡器观察到的上游状态，不是对后端全部客户端的全知统计。通过另一个代理或内部直连到来的工作，可能不在当前视野中。在 nginx 中，worker 范围和上游共享区域配置影响哪些运行状态共享；独立均衡器也不会自动合并观察，形成一个全局最少连接决定。忽略计数范围，会让局部正确的选择被误解成整个集群的最优分配。

权重使只比较原始计数不够准确。权重二的节点可以合理承载比权重一节点更多的活动连接。修改权重表达一种容量假设，并没有测量该假设是否真实。空闲 keepalive 缓存又是另一种数量，不应与这里参与选择的活动连接混淆。应分别识别正在占用的交换、为复用保留的空闲连接，以及真正消耗计算资源的工作。

快速验证

本轮复核没有重新执行 Docker 集成；保持请求与加权落点结果仍属于此前记录的观察。

将脚本保存为 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))

第一个请求进入同权最少连接组，并在某个后端处理器中等待。控制端先观察恰好一个活动请求，并识别被占用节点，然后才发送其他请求。初始平局究竟由哪台节点胜出不是主要断言；实验从后端真实状态发现该身份，不把某个偶然的初始顺序写成算法必须遵守的规则。

随后，六个短请求通过同一上游组完成。在第一个请求仍被保持时，它们全部选择另一个节点。之后控制端才放行原响应，并验证它在原节点完成。这展示的是基于持续占用的选择，而不是只统计每台服务器历史上回答过多少次请求。初始请求的释放时机受到明确控制，所以后续分配与活动状态之间的关系可以核对。

第二个上游组把 A 权重设为二、B 设为一。三个请求逐个开始，每个都保持未完成，并确认到达后才启动下一个。活动快照最终为 A=2、B=1，之后三个响应全部放行并成功完成。这说明权重会改变有效比较，策略不是永远选择原始数字最小的节点。观察快照也比仅看最终完成顺序更能说明加权占用。

实验中的 HTTP 交换让每条选中上游连接承载一个活动请求。两个节点端点共享一个可丢弃容器，因此它们的名称不代表不同硬件容量。受控开关证明的是占用与落点，不产生 CPU 基准，也不验证生产机器应该使用某个特定权重。实际节点能承受多少并行操作，仍需根据真实业务成本测量。

比较观测指标与需要保护的资源

先检查真实链路中是什么让上游连接保持活动。关闭响应缓冲时，慢客户端可能让它占用更久；开启缓冲后，代理有时能在客户端下载完成前就读完上游。因此，调整缓冲可能改变最少连接看到的计数，却没有改变应用本身的计算成本。如果忽略这层关系，就容易把连接数变化误当成应用处理能力变化。

应同时测量每个节点的延迟、错误、排队和资源使用。如果某个节点因为依赖失败而变慢，减少新连接可能缓解压力，却不会修复它。节点健康资格与重试策略仍需单独决定。最少连接不是主动健康探测，也不是让恢复节点逐渐回到原权重的升温机制；不能用一个选择算法代替这些不同控制。

最后，验证实际实现的范围和协议。其他产品也叫最少连接的方法，可能具有不同权重、持久性和状态共享规则。通过真实均衡器测试保持请求、释放和加权竞争，再用结果说明所选指标为何适合当前工作负载。算法名称本身不是通用容量保证，只有指标、作用范围与真实瓶颈对应时，选择策略才有可靠依据。

延伸阅读

来源