微缓存让可复用响应保持很短的新鲜度寿命，常见尺度是几秒。在这个窗口里，请求同一选中表示的后续访问可以复用新鲜条目。响应可能落后于源站刚发生的更新，但不能仅因为它来自缓存，就称它在 HTTP 意义上已经过期。何时过期、是否允许提供过期内容、如何协调并发填充，是三个不同策略，需要分别定义。
坏掉的样子
TTL 短并不意味着共享响应对所有用户都安全。如果个性化请求读取或填充通用缓存条目，一个用户的表示就可能在另一个上下文里被交付。proxy_cache_bypass 根据条件禁止从缓存读取；proxy_no_cache 根据条件禁止保存响应。只控制其中一侧，可能让另一条路径不符合预期隐私边界。读取已有公开条目和保存新私有响应都需要考虑。
缓存键决定哪些请求共享响应。忽略有意义的查询参数或主机维度，可能合并不同表示；把任意变化数据全部放进键，又可能让复用失效。下面实验只有一个后端 authority，并保留包含参数的完整请求 URI。真实服务如果按租户、语言或其他输入改变内容，就必须确定自己的表示身份，不能把示例键当成适用于所有应用的通用公式。
仅仅启用缓存，也不保证冷启动突发只产生一次回源。在任何响应保存完成前，多个请求可能同时未命中同一条目。proxy_cache_lock 负责协调新条目的填充，并有有限的超时和年龄控制。这个锁不是 TTL，缩短新鲜度寿命也不会隐式开启锁。新鲜条目复用与并发未命中合并，需要通过不同对照才能验证。
快速验证
下文集成实验的结果保留自历史运行记录；本次修订没有重新执行该实验，不能将这些记录当成本次环境的新实测。
将脚本保存为 check-microcache.py，准备 Python、Docker 和固定镜像后执行。记录环境为 Linux arm64 上的 nginx 1.28.0。可丢弃后端报告每条路径的请求计数，nginx 使用两秒有效期，并显式设置 Cookie 与 Authorization 绕过条件。监听仅发布到宿主回环，使用的会话和授权值都是公开测试标记，不连接真实账户：
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):
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')
docker('run','-d','--name',containers[0],'--network',name,'-p','127.0.0.1::8080',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
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'
from concurrent.futures import ThreadPoolExecutor
import threading
backend=r'''
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from urllib.parse import urlsplit,parse_qs
import json,threading
lock=threading.Lock();counts={};gates={'/lock':threading.Event(),'/nolock':threading.Event()}
class Handler(BaseHTTPRequestHandler):
protocol_version='HTTP/1.1'
def log_message(self,*args):pass
def do_GET(self):
parsed=urlsplit(self.path)
if parsed.path=='/control/state':
with lock:data=dict(counts)
elif parsed.path=='/control/release':
gates[parse_qs(parsed.query)['path'][0]].set();data={'released':True}
else:
with lock:counts[self.path]=counts.get(self.path,0)+1;count=counts[self.path]
if parsed.path in gates:assert gates[parsed.path].wait(15)
data={'path':self.path,'count':count,'cookie':self.headers.get('Cookie'),'authorization_present':bool(self.headers.get('Authorization'))}
body=json.dumps(data,sort_keys=True).encode();self.send_response(200)
if parsed.path=='/cookie':self.send_header('Set-Cookie','fixture=public-example; Path=/')
if parsed.path=='/private':self.send_header('Cache-Control','private')
if parsed.path=='/no-store':self.send_header('Cache-Control','no-store')
self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)
ThreadingHTTPServer(('0.0.0.0',8081),Handler).serve_forever()
'''
extra='''
proxy_cache_path /tmp/cache levels=1:2 keys_zone=pages:1m inactive=30s max_size=10m;
map $http_cookie $skip_cookie { default 1; "" 0; }
map $http_authorization $skip_auth { default 1; "" 0; }
proxy_cache pages; proxy_cache_key "$scheme|$proxy_host|$request_uri"; proxy_cache_valid 200 2s;
proxy_cache_bypass $skip_cookie $skip_auth; proxy_no_cache $skip_cookie $skip_auth;
proxy_cache_lock on; proxy_cache_lock_timeout 5s; proxy_cache_lock_age 5s;
add_header X-Cache $upstream_cache_status always;
'''
locations='''
location /control/ { proxy_cache off; proxy_pass http://backend:8081; }
location = /nolock { proxy_cache_lock off; proxy_pass http://backend:8081; }
location / { proxy_pass http://backend:8081; }
'''
with lab(config(locations,extra),backend=backend) as (_,request):
def get(path,headers=None):
status,h,b=request(path,headers=headers);assert status==200;return h.get('x-cache'),json.loads(b)
results={}
first=get('/item?x=1');second=get('/item?x=1');assert first[0]=='MISS' and second==('HIT',first[1])
other=get('/item?x=2');assert other[0]=='MISS' and other[1]['path']!=first[1]['path']
results['fresh']=[first[0],second[0]];results['different_query']=other[0]
for name,headers in [('cookie',{'Cookie':'session=public-fixture'}),('authorization',{'Authorization':'Bearer public-fixture'})]:
private=get('/item?x=1',headers);assert private[0]=='BYPASS' and private[1]!=first[1]
assert get('/item?x=1')==('HIT',first[1])
new='/new-'+name;assert get(new,headers)[0]=='BYPASS'
public=get(new);assert public[0]=='MISS' and public[1]['cookie'] is None and not public[1]['authorization_present']
results[name]={'warm_key':'BYPASS','public_after_private_miss':public[0]}
for path in ['/cookie','private','no-store']:
path='/'+path.lstrip('/');a=get(path);b=get(path)
assert a[0]==b[0]=='MISS' and a[1]['count']==1 and b[1]['count']==2
results[path]=[a[0],b[0]]
deadline=time.monotonic()+6
while time.monotonic()<deadline:
value=get('/item?x=2')
if value[0]=='EXPIRED':break
assert value[0]=='HIT';time.sleep(.15)
else:raise AssertionError('TTL did not expire')
assert value[1]['count']==2;results['after_ttl']={'status':value[0],'count':value[1]['count']}
for path,number,origin_count in [('/lock',6,1),('/nolock',4,4)]:
barrier=threading.Barrier(number+1)
def client():barrier.wait();return get(path)
with ThreadPoolExecutor(max_workers=number) as pool:
futures=[pool.submit(client) for _ in range(number)];barrier.wait()
wait(lambda:get('/control/state')[1].get(path)==origin_count)
time.sleep(.3);before=get('/control/state')[1][path]
assert before==origin_count and not any(f.done() for f in futures)
get('/control/release?path='+path);responses=[f.result(timeout=10) for f in futures]
assert get('/control/state')[1][path]==origin_count
statuses=sorted(r[0] for r in responses)
if path=='/lock':assert statuses==['HIT']*5+['MISS'] and all(r[1]==responses[0][1] for r in responses)
else:assert statuses==['MISS']*4
results[path]={'origin_requests_while_held':before,'cache_statuses':statuses}
print(json.dumps(results,sort_keys=True))
首次公开请求 /item?x=1 报告 MISS，下一次报告 HIT，并保留相同后端结果。/item?x=2 有自己的 MISS，响应中的路径也不同。持续检查第二个条目，最终观察到 EXPIRED，后端计数变为二。这依据服务器时钟观察实际过期，不假定精确到亚秒的切换时刻；测试关注状态变化及重新计算，而不是某个固定睡眠值。
携带示例会话 Cookie 或授权标记时，即使公开条目已经预热，请求仍报告 BYPASS。之后的公开请求仍返回原公开 HIT。另一个对照先用这些头访问尚未出现的键，再公开访问；公开请求结果是 MISS，证明之前私有上下文的响应没有填充可复用条目。这样同时覆盖了绕过读取和禁止保存，而不只是证明某次私有请求到达后端。
后端带有 Set-Cookie、Cache-Control: private 或 Cache-Control: no-store 的响应，各自连续两次都报告 MISS，源站计数也增加。配置没有忽略这些响应控制。这区分了客户端上下文触发的绕过策略和源站声明的存储限制；不能以缓存只保留几秒为理由，覆盖其中任何一种隐私信号。是否允许共享应先于寿命长短判断。
对于开启锁的冷条目，六个客户端同时开始，后端故意暂不返回响应。放行前后端只看到一次请求；放行后，一个客户端报告 MISS，五个报告 HIT，并得到相同结果。对于另一个关闭锁的条目，四个被保持的客户端产生四次回源，全部报告 MISS。受控放行使重叠的未命中直接可见，而不是从总耗时推测是否发生请求合并。
分别决定新鲜度与过载行为
本实验的两秒设置用于符合条件的 200 响应，相关源站缓存头可以优先于配置有效期。应先决定哪些响应能共享，再选择寿命，并让缓存键对应真正改变内容的输入。高命中率不能证明选中的表示正确；如果键把不同用户或不同内容合并，命中率甚至可能很好，但语义已经错误。验证内容身份和减少回源需要一起进行。
等待锁的请求用额外等待换取较少源站工作。本实验在锁限制内放行后端，并没有证明限制到期或后端失败时仍永远只有一次回源。应结合服务可接受延迟和失败行为，审阅锁等待超时、锁年龄以及可能采用的过期响应策略。是否允许返回过期条目，不能由已经设置短 TTL 这一事实代替决定。
观察缓存状态时，同时核对源站计数和响应身份。MISS 表示请求表示没有作为缓存命中提供，BYPASS 则标明这里显式排除读取的路径。新鲜 HIT、EXPIRED 后重新填充，以及有意提供过期内容，不应混成一个笼统的“已缓存”结果。它们对正确性与后端负载的影响不同，正是需要分别测量和解释的原因。
延伸阅读
来源
- nginx proxy module，缓存有效期、缓存锁、绕过读取、禁止保存与响应头控制，已核验 2026-10-15。
- nginx content caching guide，缓存键、状态观察与缓存行为，已核验 2026-10-15。
- RFC 9111 HTTP Caching，新鲜与过期响应，以及 private、no-store 指令，已核验 2026-10-15。
