缓存击穿、雪崩和穿透问题的解决方案
2023-08-296.1k 阅读
缓存击穿问题及解决方案
缓存击穿概述
缓存击穿是指在高并发场景下，一个热点 key 正好失效的瞬间，大量的请求同时访问该 key，由于缓存中不存在该 key，这些请求就会直接穿透到数据库，给数据库带来巨大压力，甚至可能导致数据库崩溃。这种情况就好像在缓存的屏障上击穿了一个洞，让大量请求得以直接通过。
解决方案
- 使用互斥锁（Mutex）
- 原理：在缓存失效时，不是立即去查询数据库，而是先尝试获取一个互斥锁。只有获取到互斥锁的请求才能去查询数据库并更新缓存，其他请求则等待。当获取锁的请求更新完缓存后，释放互斥锁，其他等待的请求可以直接从缓存中获取数据。
- 代码示例（以 Python + Redis 为例）
import redis
import time
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
def get_data_with_mutex(key):
data = redis_client.get(key)
if data is None:
mutex_key = f'mutex:{key}'
# 尝试获取互斥锁
acquired = redis_client.set(mutex_key, 'lock', nx=True, ex=10)
if acquired:
try:
# 从数据库查询数据
data = get_data_from_db(key)
if data:
redis_client.set(key, data)
return data
finally:
# 释放互斥锁
redis_client.delete(mutex_key)
else:
# 未获取到锁，等待一段时间后重试
time.sleep(0.1)
return get_data_with_mutex(key)
return data.decode('utf-8')
def get_data_from_db(key):
# 模拟从数据库查询数据
return f'data for {key}'
- 设置热点数据永不过期
- 原理：对于热点数据，不设置过期时间，这样就不会出现缓存失效的情况，也就避免了缓存击穿。但需要注意的是，虽然数据永不过期，但在数据发生变化时，需要及时更新缓存。
- 代码示例（以 Java + Redis 为例）
import redis.clients.jedis.Jedis;
public class HotDataNeverExpire {
private static Jedis jedis = new Jedis("localhost", 6379);
public static String getData(String key) {
String data = jedis.get(key);
if (data == null) {
// 从数据库查询数据
data = getDataFromDB(key);
if (data != null) {
// 设置热点数据永不过期
jedis.set(key, data);
}
}
return data;
}
private static String getDataFromDB(String key) {
// 模拟从数据库查询数据
return "data for " + key;
}
}
- 基于二级缓存
- 原理：设置两个缓存，一级缓存为短时间过期的缓存，二级缓存为长时间过期或永不过期的缓存。当一级缓存失效时，先从二级缓存获取数据，如果二级缓存也没有，则查询数据库并同时更新一级和二级缓存。
- 代码示例（以 Go + Redis 为例）
package main
import (
"fmt"
"github.com/go-redis/redis/v8"
"context"
"time"
)
var ctx = context.Background()
var rdb *redis.Client
func init() {
rdb = redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "",
DB: 0,
})
}
func getDataWithDoubleCache(key string) string {
data, err := rdb.Get(ctx, key).Result()
if err == redis.Nil {
// 一级缓存失效，从二级缓存获取
secData, err := rdb.Get(ctx, fmt.Sprintf("sec:%s", key)).Result()
if err == redis.Nil {
// 二级缓存也失效，查询数据库
data = getDataFromDB(key)
if data != "" {
// 更新一级和二级缓存
rdb.Set(ctx, key, data, 5*time.Minute)
rdb.Set(ctx, fmt.Sprintf("sec:%s", key), data, 24*time.Hour)
}
} else if err == nil {
data = secData
}
} else if err != nil {
panic(err)
}
return data
}
func getDataFromDB(key string) string {
// 模拟从数据库查询数据
return fmt.Sprintf("data for %s", key)
}
缓存雪崩问题及解决方案
缓存雪崩概述
缓存雪崩是指在某一时刻，大量的缓存数据同时过期失效，导致大量请求直接访问数据库，使得数据库压力骤增，甚至可能造成数据库服务崩溃。与缓存击穿不同，缓存击穿是单个热点 key 失效，而缓存雪崩是大量 key 同时失效。
解决方案
- 随机设置过期时间
- 原理：在设置缓存过期时间时，不使用固定的过期时间，而是设置一个随机的过期时间范围。这样可以避免大量缓存同时过期，将缓存过期的时间分散开，降低数据库瞬间承受高并发的风险。
- 代码示例（以 Python + Redis 为例）
import redis
import random
import time
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
def set_data_with_random_expiry(key, value):
min_expiry = 3600 # 最小过期时间，单位秒
max_expiry = 7200 # 最大过期时间，单位秒
expiry = random.randint(min_expiry, max_expiry)
redis_client.setex(key, expiry, value)
def get_data(key):
data = redis_client.get(key)
if data is None:
# 从数据库查询数据
data = get_data_from_db(key)
if data:
set_data_with_random_expiry(key, data)
return data
return data.decode('utf-8')
def get_data_from_db(key):
# 模拟从数据库查询数据
return f'data for {key}'
- 使用二级缓存
- 原理：和缓存击穿解决方案中的二级缓存类似，一级缓存设置较短的过期时间，二级缓存设置较长的过期时间。当一级缓存失效时，先从二级缓存获取数据，减轻数据库的压力。
- 代码示例（以 Node.js + Redis 为例）
const redis = require('redis');
const { promisify } = require('util');
const client = redis.createClient(6379, 'localhost');
const getAsync = promisify(client.get).bind(client);
const setexAsync = promisify(client.setex).bind(client);
async function getDataWithDoubleCache(key) {
let data = await getAsync(key);
if (data === null) {
let secData = await getAsync(`sec:${key}`);
if (secData === null) {
data = await getDataFromDB(key);
if (data) {
await setexAsync(key, 300, data);
await setexAsync(`sec:${key}`, 3600, data);
}
} else {
data = secData;
}
}
return data;
}
async function getDataFromDB(key) {
// 模拟从数据库查询数据
return `data for ${key}`;
}
- 缓存预热
- 原理：在系统上线前或者业务低峰期，提前将一些热点数据加载到缓存中，并设置合理的过期时间。这样在高峰期来临时，缓存中有足够的数据，减少了缓存雪崩的可能性。
- 代码示例（以 C# + Redis 为例）
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace CachePreheating
{
class Program
{
private static ConnectionMultiplexer redis = ConnectionMultiplexer.Connect("localhost:6379");
private static IDatabase db = redis.GetDatabase();
static void Main(string[] args)
{
// 模拟缓存预热
List<string> hotKeys = new List<string> { "key1", "key2", "key3" };
foreach (var key in hotKeys)
{
string data = GetDataFromDB(key);
db.StringSet(key, data, TimeSpan.FromHours(1));
}
}
static string GetDataFromDB(string key)
{
// 模拟从数据库查询数据
return $"data for {key}";
}
}
}
- 熔断降级
- 原理：当发现数据库压力过大时，暂时停止从缓存失效后查询数据库的操作，直接返回一个默认值或者提示信息，避免数据库被压垮。等数据库压力恢复后，再恢复正常的查询流程。
- 代码示例（以 Java + Hystrix 为例，Hystrix 是一个实现熔断降级的框架）
import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import redis.clients.jedis.Jedis;
public class FallbackExample {
private static Jedis jedis = new Jedis("localhost", 6379);
public static String getData(String key) {
return new HystrixCommand<String>(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")) {
@Override
protected String run() throws Exception {
String data = jedis.get(key);
if (data == null) {
data = getDataFromDB(key);
if (data != null) {
jedis.set(key, data);
}
}
return data;
}
@Override
protected String getFallback() {
// 熔断降级，返回默认值
return "default data";
}
}.execute();
}
private static String getDataFromDB(String key) {
// 模拟从数据库查询数据
return "data for " + key;
}
}
缓存穿透问题及解决方案
缓存穿透概述
缓存穿透是指查询一个根本不存在的数据，由于缓存中没有，每次请求都会去查询数据库，而数据库中也不存在该数据，导致大量无效请求穿透缓存直接访问数据库，消耗数据库资源。这种情况就好像缓存对这些请求完全没有起到阻挡作用，如同穿透了一般。
解决方案
- 布隆过滤器（Bloom Filter）
- 原理：布隆过滤器是一种概率型数据结构，它可以用来判断一个元素是否在一个集合中。它的优点是空间效率和查询时间都非常好，缺点是存在一定的误判率。在缓存穿透场景中，使用布隆过滤器先判断请求的数据是否存在，如果不存在则直接返回，不会去查询数据库，从而避免无效请求对数据库的压力。
- 代码示例（以 Python + BloomFilter 库为例，需要先安装
pybloomfiltermmap库）
from pybloomfiltermmap import BloomFilter
import redis
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
# 假设预计有10000个元素，误判率为0.01
bloom = BloomFilter(capacity=10000, error_rate=0.01, filename='bloomfilter.dat')
def get_data_with_bloom(key):
if key not in bloom:
return None
data = redis_client.get(key)
if data is None:
# 从数据库查询数据
data = get_data_from_db(key)
if data:
redis_client.set(key, data)
bloom.add(key)
return data
return data.decode('utf-8')
def get_data_from_db(key):
# 模拟从数据库查询数据
return f'data for {key}'
- 空值缓存
- 原理：当查询数据库发现数据不存在时，也将这个空值缓存起来，并设置一个较短的过期时间。这样下次再查询同样不存在的数据时，直接从缓存中获取空值，而不会去查询数据库，减少了无效请求对数据库的压力。
- 代码示例（以 PHP + Redis 为例）
<?php
$redis = new Redis();
$redis->connect('localhost', 6379);
function getDataWithNullCache($key)
{
global $redis;
$data = $redis->get($key);
if ($data === false) {
$data = getDataFromDB($key);
if ($data === null) {
// 缓存空值，设置较短过期时间
$redis->setex($key, 60, '');
return null;
} else {
$redis->set($key, $data);
return $data;
}
}
return $data;
}
function getDataFromDB($key)
{
// 模拟从数据库查询数据
if ($key === 'existing_key') {
return 'data for existing_key';
}
return null;
}
- 参数校验
- 原理：在接收到请求时，对请求参数进行严格校验，确保参数的合法性和合理性。对于明显不合理或者不存在的数据请求，直接返回错误信息，不进行后续的缓存和数据库查询操作。
- 代码示例（以 Python Flask 框架为例）
from flask import Flask, request
app = Flask(__name__)
@app.route('/data')
def get_data():
key = request.args.get('key')
if not key or key.startswith('invalid_'):
return 'Invalid request', 400
# 这里可以继续进行缓存和数据库查询操作
return 'Data retrieval logic'
if __name__ == '__main__':
app.run(debug=True)