一、先来说说这个场景
大家每天用手机点外卖、刷新闻、看商品，后端服务都在疯狂做同一件事：把数据拿出来展示给用户。这些数据大部分是读操作，比如一个商品详情被访问一万次，可能只有一次是修改价格。这种“读多写少”的场景，我们叫它“读密集”。
我打个比方。DynamoDB 就像一家超级大仓库，东西放得整整齐齐，但每次去仓库取货都要走很长的路，而且仓库管理员按访问次数收费。如果每个用户来买东西都直接往仓库跑，仓库不仅累，账单也会吓死人。这时候，我们需要在前台设一个“小货柜”，这就是 ElastiCache。小货柜里放着最常卖的商品，用户来了直接从小货柜拿，不到万不得已不去仓库。
但这里有一个绕不开的坑：如果用户来问一件仓库里压根没有的东西，小货柜里自然也没有。你每次都放他去仓库找，仓库翻个底朝天也找不到，这种无谓的折腾叫“缓存穿透”。今天我们就聊聊怎么把这两套东西配合好，既挡住穿透，又把成本省下来。
二、先搭一个最基础的缓存读写
2.1 最简单的代码长什么样
假设我们有个商品表，主键是商品 ID。用 Python 写的话，最直接的逻辑是：先去 Redis 查，查到就返回；查不到就去 DynamoDB 找，找到后塞回 Redis。代码如下：
# 技术栈：Python + boto3（AWS SDK）
import boto3
import redis
# 创建 DynamoDB 客户端
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('products')
# 创建 Redis 客户端（ElastiCache Redis 的地址）
cache = redis.Redis(
host='mycache.xxxxx.ng.0001.use1.cache.amazonaws.com',
port=6379,
decode_responses=True
)
def get_product(product_id):
# 先从缓存读取
cached = cache.get(f'product:{product_id}')
if cached is not None:
return cached # 缓存命中，直接返回
# 缓存没命中，去 DynamoDB 查
response = table.get_item(Key={'id': product_id})
item = response.get('Item')
# 查到了，写回缓存，设置 10 分钟过期，防止长时间不更新
if item:
import json
cache.setex(f'product:{product_id}', 600, json.dumps(item))
return item
# 查不到就返回空
return None
这个方案看起来没问题，但要是用户疯狂查询一个根本不存在的 ID，比如 product:100000，Redis 里查不到，每次都去 DynamoDB 查，DynamoDB 就得白白挨打。更糟的是，攻击者可以故意用不存在的 ID 刷接口，让数据库承受巨大的读压力，这就是“缓存穿透”的典型症状。
2.2 穿透到底有多疼
DynamoDB 的按需计费按照读取容量单位收费，每次查询即使找不到数据，也会消耗读能力。读密集场景下，如果 10% 的请求都打在空数据上，那这 10% 的成本就纯属浪费。而且，DynamoDB 有配额限制，大量无效查询会把正常的业务查询挤掉，导致真实用户的请求被限流。
所以，我们必须想办法让那些“空白查询”不要真的冲到数据库去。
三、防穿透的常用办法
3.1 缓存空值：最直观的解决方案
既然空结果也会被反复查，那我们就把它也缓存起来。只不过空值的过期时间要短一点，比如 1 分钟，避免占用太多内存。代码改动不大：
# 技术栈：Python + boto3（AWS SDK）
import json
import boto3
import redis
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('products')
cache = redis.Redis(
host='mycache.xxxxx.ng.0001.use1.cache.amazonaws.com',
port=6379,
decode_responses=True
)
def get_product_with_null_cache(product_id):
# 先查缓存，注意：不管有没有值，只要 key 存在就直接返回
cached = cache.get(f'product:{product_id}')
if cached is not None:
# 空值的话，缓存里存的是特殊标记，我们约定用 "NULL" 表示
if cached == 'NULL':
return None
return json.loads(cached)
# 缓存没有，去 DynamoDB 查
response = table.get_item(Key={'id': product_id})
item = response.get('Item')
if item:
# 真实数据缓存 10 分钟
cache.setex(f'product:{product_id}', 600, json.dumps(item))
return item
# 查不到，缓存空值 60 秒，防止穿透
cache.setex(f'product:{product_id}', 60, 'NULL')
return None
这种方案的好处是简单，容易理解，对突然涌进来的无效请求效果立竿见影。缺点是如果恶意请求的 key 一直在变，比如每次拼一个随机数，那缓存里会堆积大量空值 key，白白消耗内存。所以还需要结合后面的布隆过滤器，或者对空值 key 做更短的有效期。
3.2 布隆过滤器：从源头拦截
布隆过滤器是一个很巧妙的东西。它用很省内存的方式表达“某个 key 大概率存在”。我们可以把所有商品 ID 都放进去，查询前先问布隆过滤器：“这个 ID 存在吗？”如果它说“不存在”，那肯定不存在，直接返回空，连缓存和数据库都不碰。如果它说“存在”，才继续走缓存和数据库。
注意，布隆过滤器有误判率，它可能会把不存在的 ID 误判为存在，但永远不会把存在的 ID 误判为不存在。所以它是一个极好的“前置筛子”。
用 Python 手写一个简单的布隆过滤器也不难：
# 技术栈：Python 内置实现（基于位数组，仅用于演示原理）
import math
import mmh3 # 需要安装 mmh3，这里只是展示思路
class SimpleBloomFilter:
def __init__(self, expected_items, error_rate=0.01):
# 根据预计元素数量和误判率计算位数组大小
self.size = int(-expected_items * math.log(error_rate) / (math.log(2) ** 2))
# 需要几个哈希函数
self.hash_count = int(self.size * math.log(2) / expected_items)
self.bit_array = [0] * self.size
def add(self, item):
# 对同一个 item 算多个哈希，把对应的位都置为 1
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
self.bit_array[index] = 1
def contains(self, item):
# 只要有一个位是 0，就说明一定不存在
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
if self.bit_array[index] == 0:
return False
return True
实际使用时，可以把所有商品 ID 在后台任务里提前导入到布隆过滤器，或者用 Redis 模块的方式直接在 Redis 里维护。在 DynamoDB 的场景下，如果数据量不大，也可以直接用 Lua 脚本放在 Redis 里运行。这里为了演示用了 Python 标准思路，读者可以自行替换成 Redis 的布隆过滤器类型。
有了布隆过滤器，代码逻辑变成：
# 技术栈：Python + boto3（使用上面的 SimpleBloomFilter）
import json
import boto3
import redis
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('products')
cache = redis.Redis(
host='mycache',
port=6379,
decode_responses=True
)
bloom = SimpleBloomFilter(100000) # 假设有 10 万商品
def get_product_safe(product_id):
# 第一步：布隆过滤器说没有，直接返回 None
if not bloom.contains(product_id):
return None
# 第二步：查缓存（还是可以用空值缓存）
cached = cache.get(f'product:{product_id}')
if cached is not None:
return None if cached == 'NULL' else json.loads(cached)
# 第三步：查数据库
item = table.get_item(Key={'id': product_id}).get('Item')
if item:
cache.setex(f'product:{product_id}', 600, json.dumps(item))
else:
cache.setex(f'product:{product_id}', 60, 'NULL')
return item
这样，大量不存在的 ID 会在第一步被拦截，根本到不了 DynamoDB。
3.3 限流和熔断作为保底
即使做了上面两层防护，也不能保证万无一失。比如布隆过滤器有误判，或者极端情况下缓存集体失效，大量请求还是会打向数据库。这时候需要给缓存和数据库之间加一个限流器或熔断器。我们可以用 Python 在应用层做一个简单的令牌桶，每个商品 ID 一秒钟最多查一次数据库，多余的请求直接返回一个提示。这个属于保护措施，代码就不展开了，但大家要记住：任何防护都不能让你完全躺平，兜底机制很重要。
四、降本增效：怎么把成本降下来
4.1 DynamoDB 容量模式的选择
DynamoDB 有两种计费模式：按需模式和预置模式。按需模式适合流量忽高忽低，预置模式适合相对稳定的流量。读密集场景下，如果我们的缓存命中率很高，真正打到 DynamoDB 的请求少得可怜，那么可以大胆地选择预置模式，并且把读容量设置得很低。比如商品查询接口每秒有 5000 个请求，但缓存命中率是 99%，那么 DynamoDB 每秒只需要处理 50 个查询，预置 100 个读容量就已经很宽裕了。这样每个月的账单能少一大截。
用 boto3 可以随时调整预置容量，比如在缓存初建时调高，稳定后调低：
# 技术栈：Python + boto3（AWS SDK）
import boto3
client = boto3.client('dynamodb', region_name='us-east-1')
# 把 products 表的读容量调整为 100，写容量保持 50
response = client.update_table(
TableName='products',
ProvisionedThroughput={
'ReadCapacityUnits': 100,
'WriteCapacityUnits': 50
}
)
print(response) # 更新成功
当然，如果业务流量波动很大，按需模式更省心，因为不需要预测容量。但预置模式在成本上更有优势，适合读密集且量大的稳定业务。
4.2 ElastiCache 的合理配置
ElastiCache 的计费和实例规格直接相关。我们不需要买一台超大规格的缓存集群，只需要根据实际缓存的数据量估算。假设每个商品详情序列化后 2KB，缓存 10 万个商品，大概只需要 200MB 内存，一台小规格实例就绰绰有余。TTL 设置也很关键，太短则命中率低，太长则数据更新不及时。一般业务数据可以设 5 到 15 分钟，热点数据可以更久。
另外，ElastiCache 支持集群模式，可以把 key 打散在多台机器上。但集群模式下，批量操作和某些排序命令会受限，对于简单的 key-value 读取来说，单节点加副本就够了。副本主要是为了提高高可用，不是用来扩展写性能，因为缓存层的写也很轻。
4.3 数据一致性：更新数据库后怎么同步缓存
缓存和数据同步有几种套路。比较简单的是“先更新数据库，再删除缓存”，下次查询时重新加载。在 DynamoDB 场景下，也可以使用 DynamoDB Streams 捕获数据变更，然后通过 Lambda 函数去更新或删除 Redis 对应 key。我们这里不展开 Lambda 和运维细节，只给出同步思路：
# 技术栈：Python + boto3（使用 DynamoDB Streams 模拟）
# 该代码示意在 Lambda 中处理 stream 记录
import boto3
cache = None # 实际需要初始化 Redis 连接
def handle_stream(event, context):
for record in event['Records']:
# DynamoDB Streams 记录中包含新老数据
new_image = record['dynamodb'].get('NewImage')
old_image = record['dynamodb'].get('OldImage')
if new_image:
product_id = new_image['id']['S']
elif old_image:
product_id = old_image['id']['S']
else:
continue
# 删除缓存，下次读取时自然回填
cache.delete(f'product:{product_id}')
这个逻辑简单可靠。删除缓存比更新缓存更省事，也避免并发写缓存时出现旧数据。如果要更精细，可以在数据库写入时直接更新缓存，但需要处理并发和时钟问题，不如先删后查来得干净。
4.4 缓存预热和热点 key
在系统刚上线时，Redis 里是空的，如果突然有大量用户涌入，每个请求都会穿透到 DynamoDB，我们称为“缓存雪崩”。所以要提前把热门商品的数据加载到缓存里。可以写一个脚本扫一遍商品表，把最近销售最多的前几百个商品回填到 Redis。热点 key 的另一个问题是，某个超级热门的商品会频繁被访问，虽然命中缓存，但 Redis 单节点也可能成为瓶颈。这个时候可以考虑把热点 key 复制成多份，比如在 key 后面加一个随机后缀，分散到不同分片。这个技巧在 ElastiCache 集群模式下很有用。
五、一个完整的设计方案
现在我们把这些东西拼成一个完整方案。假设我们是一个在线商城的商品服务，接口需要根据商品 ID 返回完整信息。整体流程是：
- 请求到达服务，先查布隆过滤器；
- 布隆过滤器说不存在，直接返回空；
- 布隆过滤器说存在，去 Redis 查；
- Redis 有数据（包括空值标记），直接返回；
- Redis 没有，去 DynamoDB 查；
- DynamoDB 查到，写回 Redis，TTL 10 分钟；没查到，写回空值，TTL 1 分钟。
完整代码如下：
# 技术栈：Python + boto3 + redis-py
# 这个文件是一个完整的商品查询服务示例，包含布隆过滤器和空值缓存
import json
import math
import boto3
import redis
import mmh3 # 演示用，实际可换用 Redis 布隆过滤器模块
class SimpleBloomFilter:
def __init__(self, expected_items, error_rate=0.01):
self.size = int(-expected_items * math.log(error_rate) / (math.log(2) ** 2))
self.hash_count = int(self.size * math.log(2) / expected_items)
self.bit_array = [0] * self.size
def add(self, item):
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
self.bit_array[index] = 1
def contains(self, item):
for seed in range(self.hash_count):
index = mmh3.hash(item, seed) % self.size
if self.bit_array[index] == 0:
return False
return True
# 初始化外部依赖
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('products')
cache = redis.Redis(
host='mycache.xxxxx.ng.0001.use1.cache.amazonaws.com',
port=6379,
decode_responses=True
)
# 初始化布隆过滤器，并预加载所有商品 ID
bloom = SimpleBloomFilter(100000)
def load_all_product_ids():
"""预加载商品 ID 到布隆过滤器（实际可定时任务更新）"""
scan_kwargs = {}
while True:
response = table.scan(ProjectionExpression='id', **scan_kwargs)
for item in response['Items']:
bloom.add(item['id'])
if 'LastEvaluatedKey' in response:
scan_kwargs['ExclusiveStartKey'] = response['LastEvaluatedKey']
else:
break
def get_product(product_id):
# 1. 布隆过滤器拦截不存在的 ID
if not bloom.contains(product_id):
return None
# 2. 查缓存
cache_key = f'product:{product_id}'
cached = cache.get(cache_key)
if cached is not None:
return None if cached == 'NULL' else json.loads(cached)
# 3. 查数据库
response = table.get_item(Key={'id': product_id})
item = response.get('Item')
# 4. 回填缓存
if item:
cache.setex(cache_key, 600, json.dumps(item))
else:
# 空值也缓存，60 秒后过期，让新创建的商品能尽快被访问
cache.setex(cache_key, 60, 'NULL')
return item
这份代码虽然不长，但已经把核心环节演示清楚了。实际生产里，你需要把布隆过滤器的数据做成可更新，比如每天同步一次商品表；Redis 连接要使用连接池；DynamoDB 的访问要配置好 IAM 权限和重试策略。
六、技术优缺点与注意事项
6.1 这套方案有哪些优点
第一，成本下降明显。读密集场景下，99% 的请求在 Redis 层就返回了，DynamoDB 的读容量可以降得很低，账单自然就下来了。第二，响应速度更快。从 Redis 读数据通常一毫秒左右，比 DynamoDB 动辄十几毫秒要快得多。第三，防穿透能力强。布隆过滤器加空值缓存，双保险挡住无效请求，数据库真正读到的几乎都是有效数据。
6.2 有哪些缺点
布隆过滤器增加了系统复杂度，需要维护一份“合法的 key 列表”，商品 ID 变化时要及时更新，否则新商品会被误杀。空值缓存会占用一些 Redis 内存，如果攻击者使用随机不存在的 ID，空值 key 会膨胀。另外，缓存层本身也有风险，比如 Redis 内存满了会触发淘汰策略，可能导致热点 key 被异常清除。还有，Redis 是单线程模型，如果某个慢命令阻塞了，整个缓存层都会受影响。
6.3 需要注意的细节
- 缓存 key 的设计要统一，比如都用 product:{id} 这样的格式，方便统计和排查。
- TTL 不是越大越好，也不是越小越好，要结合业务对实时性的要求。
- DynamoDB 预置模式要开启自动扩缩容，避免流量突然上涨时被打爆。
- 布隆过滤器最好放在应用内存里，或者用 Redis 官方模块，避免每个请求都经过网络额外查询。
- 要监控缓存命中率、数据库读容量消耗、Redis 内存占用等指标，及时调整参数。
6.4 适合在什么场景下用
这套组合拳特别适合下面几种情况：第一，商品详情、用户资料、配置信息这类读多写少的数据。第二，业务流量有典型的波峰波谷，比如白天高、晚上低，缓存能很好地扛住波峰。第三，数据库成本压力大，希望通过缓存降低数据库吞吐量。反之，如果业务是写多读少，或者每次查询结果都带强一致性要求，缓存反而可能造成麻烦。
七、总结
读密集场景下的缓存设计，本质是让数据离用户更近，让系统花更少的钱办更多的事。DynamoDB 负责可靠存储，ElastiCache 负责快速响应用户，两者加起来，再配上布隆过滤器和空值缓存，就能把缓存穿透的问题扼杀在摇篮里。成本上，通过预置容量和合理缓存策略，能省下真金白银。当然，技术选型没有银弹，最重要的是理解原理，然后根据业务数据特征做出取舍。如果你现在正被数据库高成本和高延迟困扰，不妨试试这套组合拳。
评论
围绕“DynamoDB与ElastiCache缓存层协同架构设计详解读密集场景下的缓存穿透防范与成本效益提升的降本增效方案”参与讨论