Redis缓存穿透-缓存击穿-缓存雪崩-布隆过滤器

缓存穿透

缓存穿透

指在redis缓存中不存在数据，这个时候只能去访问持久层数据库，当用户很多时，缓存都没有命中就会照成很大压力

解决方案

  1. 布隆过滤器（对可能查询的数据先用hash存储）
  2. 缓存空对象：在没有的数据中存一个空，而这些空的对象会设置一个有效期）

缓存击穿

缓存击穿指的是缓存中没有数据但数据库中有数据（一般是热点数据缓存时间到期），同一时间大量的并发请求由于读缓存没读到数据，就去数据库去取数据，导致某个时间内数据库压力剧增，导致崩溃。

缓存击穿的解决方案

  1. 设置热点数据永远不过期（可以判断当前key快要过期时，通过后台异步线程在重新构建缓存）
  2. 接口限流与熔断，降级。重要的接口一定要做好限流策略，防止用户恶意刷接口，同时要降级准备，当接口中的某些服务不可用时候，进行熔断，失败快速返回机制。
  3. 设置互斥锁。在并发的多个请求中，只有第一个请求线程能拿到锁并执行数据库查询操作，其他的线程拿不到锁就阻塞等着，等到第一个线程将数据写入缓存后，直接走缓存。
    • 就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db。
    • 先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX）去set一个互斥锁。
    • 当操作返回成功时，再进行load db的操作，并回设缓存,最后删除mutex key；
    • 当操作返回失败，证明有线程在load db，当前线程睡眠一段时间再重试整个get缓存的方法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
<?php

class Lock
{
	/**
	 * redis 
	 * 
	 * @var Redis 
	 */
	private $redis;

	/**
	 * id lenght
	 *
	 * @var integer
	 */
	private $randomLen = 16;
 
	/**
	 * expire tolarance
	 *
	 * @var integer
	 */	
	private $tolerance = 500; // milliseconds

	/**
	 * seconds to millis
	 *
	 * @var integer
	 */
	private $millisPerSeconds = 1000;

	/**
	 * set redis lua script 
	 *
	 * @var string
	 */
	private $lockCommand = 'if redis.call("GET", KEYS[1]) == ARGV[1] then
	redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2])
	return "OK"
else
	return redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2])
end
';

	/**
	 * get redis lua script
	 *
	 * @var string
	 */
	private $delCommand = 'if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end
';

	/**
	 * lock key
	 *
	 * @var string 
	 */
	private $key;

	/**
	 * lock id
	 *
	 * @var [type]
	 */
	private $id;

	/**
	 * lock expire 
	 *
	 * @var integer
	 */
	private $seconds = 0;

	public function __construct(Redis $redis, string $key)
	{
		$this->redis = $redis;
		$this->id = $this->randomUUID();
		$this->key = $key;
	}

	/**
	 * generate uuid 
	 * provisional way
	 *
	 * @return string
	 */
	public function randomUUID(): string
	{
		return substr(uniqid('', true), 5, $this->randomLen);
	}

	/**
	 * expire
	 *
	 * @return integer
	 */
	public function getPexpire(): int
	{
		return $this->seconds * $this->millisPerSeconds + $this->tolerance;
	}

	/**
	 * acquire lock 
	 *
	 * @return boolean
	 */
	public function acquire(): bool
	{
		$resp = $this->redis->eval($this->lockCommand, [$this->key, $this->id, $this->getPexpire()], 1);
		if ($resp == 'OK') {
			return true;
		}

		return false;
	}	

	/**
	 * release lock 
	 *
	 * @return boolean
	 */
	public function release(): bool
	{
		$resp = $this->redis->eval($this->delCommand, [$this->key, $this->id], 1);
	
		return $resp == 1;
	}

	/**
	 * set expire
	 *
	 * @param integer $expire
	 * @return void
	 */
	public function setExpire(int $expire)
	{
		$this->seconds = $expire;
	}
}

// test
$redis = new Redis();
$redis->connect('192.168.4.61', 6379);
$key = 'lock:testtest';
$lock = new Lock($redis, $key);
$lock->setExpire(60);
var_dump($lock->acquire());
var_dump($lock->acquire());
// var_dump($lock->release());
$lock2 = new Lock($redis, $key);
$lock2->setExpire(30);
var_dump($lock2->acquire());

<?php
class Cache
{
	/**
	 * redis
	 *
	 * @var Redis 
	 */
	private $redis;

	/**
	 * wait next get
	 * milliseconds
	 * 
	 * @var integer
	 */
	private $rate = 200;

	/**
	 * max loops
	 *
	 * @var integer
	 */
	private $maxLoops = 5;

	public function __construct($redis)
	{
		$this->redis = $redis;
	}

	public function setRate(int $rate)
	{
		$this->rate = $rate;
	}

	/**
	 * 第一版,初步实现
	 * 通过闭包的方式实现
	 *
	 * @param string $key
	 * @param Closure $fn
	 * @return mixed
	 */
	public function getCache($key, Closure $fn)
	{
		for ($i = 0; $i < $this->maxLoops; $i ++) {
			// get redis
			$value = $this->redis->get($key);
			if (!empty($value)) {
				return $value;
			}
			// global lock
			$lock = new Lock();
			if (!$lock->lock()) {
				// wait next get
				usleep($this->rate * 1000);
				continue;
			}
			$value = $fn();	
			// 写入Redis
			$this->redis->set($key, $value);
			// global unlock	
			$lock->unlock();
	
			return $value;
	
		}

		return 0;
	}
}

// DB Model
class User
{
	public function getName()
	{
		return 'jerry';
	}

	public function getUserById($userId)
	{
		$key = 'cache:user_id' . $userId;
		$redis = new Redis();
		$redis->connect('192.168.4.61', 6379);
		$cache = new Cache($redis);
		$cache->setRate(300);
		return $cache->getCache($key, function () use ($userId) {
			// DB
			return (new User())->getName();	
		});
	}
}

// test
$user = new User();
echo $user->getUserById(1);

缓存雪崩

缓存雪崩：在某个时间段，缓存集体过期、redis宕机

解决方案

给key的失效时间设置为随机时间，避免集体过期；双缓存；加互斥锁

解决方式

  1. 若是由于大量key过期所造成的，可以给key的ttl设置随机时间，避免集体过期
  2. 若是因为redis服务器宕机所导致的，可以搭建redis集群，保证高可用
  3. 可以从请求量层面进行解决，对缓存业务添加限流和服务降级策略
  4. 可以添加多级缓存，比如说nginx缓存

布隆过滤器

我们可以这样考虑，可以先判断key值是否存在，如果不存在，则不访问redis，那这样就可以拦截大量的请求，布隆过滤器恰好可以实现这样的需求。

布隆过滤器本质是一个二进制向量，初始化的时候每一个位置都是0，如下图，比如说a经过hash算法后得到一个下标位置，接下来就会把下标的值改为1，图中所示的是每一个元素经过三次hash运算，每一个红线代表一次hash算法，为什么要运算三次呢，这是为了减少hash冲突，当然hash算法不一定是三次，经过多次不同维度的哈市算法后，就把a值映射到了二进制向量里面，这样的好处很多，可以节省空间，假如说a值是一串很长的字符串，那么经过映射后就可以只占三位长度，并且查找速度很快。

如果布隆过滤器判断元素存在，则不一定存在，如果不存在，则一定不存在
如何理解这句话，因为有可能你一个元素运算得到的下标恰好是别的元素的下标，如果经过运算后布隆过滤器判断不存在，也就是说至少有一个下标是为0的，那肯定是不存在的

布隆过滤器的使用

注意的是布隆过滤器有一定的误判率，不可能达到100%的精准,首先初始化项目的时候从数据库查询出来所有的key值,然后放到布隆过滤器中。

实现代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
<?php
Class Bloom
{
	/* $var int $maps */
	private $maps = 14;	

	/* $var int $bits */
	
	private $bits = 20000;

	/* $var string $setScript */
	private $setScript = '
for _, offset in ipairs(ARGV) do
	redis.call("setbit", KEYS[1], offset, 1)
end
return true
';
	
	/* $var string $testScript */
	private $testScript = '
for _, offset in ipairs(ARGV) do
	if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
		return false
	end
end
return true
';

	/* $var string $key*/
	private $key;	
	
	/* $var mixed $redis*/
	private $redis;

	/**
	 * construct
	 * 
	 * @param mixed $redis
	 * @param string $key
	 * @return void
	 */
	public function __construct($redis, $key)
	{
		$this->redis = $redis;	
		$this->key = $key;
	}
	
	/**
	 * 格式化参数
	 *
	 * @param array $offsets
	 * @return array
	 */
	private function buildOffsetArgs(array $offsets): array
	{
		array_unshift($offsets, $this->key);
		return $offsets;
	}

	/**
	 * 设置过滤器的值
	 *
	 * @param array $offsets
	 * @return boolean
	 */
	private function set(array $offsets): bool
	{
		return $this->redis->eval($this->setScript, $this->buildOffsetArgs($offsets), 1);
	}

	/**
	 * 检查值是否存在
	 *
	 * @param array $offsets
	 * @return boolean
	 */
	private function check(array $offsets): bool
	{
		return $this->redis->eval($this->testScript, $this->buildOffsetArgs($offsets), 1);
	}

	/**
	 * 计算bloom过滤器对应的bit位
	 * 暂时使用crc32 获取摘要
	 *
	 * @param string $data
	 * @return array
	 */
	private function getLocations(string $data): array
	{
		$locations = [];
		for ($i = 0; $i < $this->maps; $i ++) {
			$data .= (string) $i;
			$hashValue = crc32($data);
			$locations[$i] = $hashValue % $this->bits;
		}

		return $locations;
	}

	/**
	 * 增加过滤器数据
	 *
	 * @param string $data
	 * @return void
	 */
	public function add(string $data)
	{
		$locations = $this->getLocations($data);
		
		return $this->set($locations);
	}

	/**
	 * 判断值是否存在
	 *
	 * @param string $data
	 * @return boolean
	 */
	public function exists(string $data): bool
	{
		$locations = $this->getLocations($data);
		
		return $this->check($locations);
	}
	
	/**
	 * 设置时间 
	 *
	 * @param int $second
	 * @return boolean
	 */

	public function expire(int $second): bool
	{
		return $this->redis->expire($this->key, $second);
	}
}

// 测试
$redis = new Redis();
$redis->connect('192.168.4.61', 6379);

$bloomKey = 'bloom:test';
$data = 'helloworld2';
$bloom = new Bloom($redis, $bloomKey);
// var_dump($bloom->add($data));
// var_dump($bloom->expire(600));
var_dump($bloom->exists($data));

加了布隆过滤器的过程如下

  1. 当应用访问的时候，先去布隆过滤器中判断kedy值，如果发觉没有key值不存在，直接返回
  2. 如果key值在布隆过滤器存在，则去访问redis，由于是有误判率的，所以redis也有可能不存在
  3. 那么这时候就去访问数据库，数据库不存在，那就直接返回空就行

如果误判率为3%，当有100万个请求同时过来的时候，布隆过滤器已经挡住了97万个请求，剩下3万个请求假如是误判的，这时候再访问数据库可以通过加锁的方式实现，只有竞争到锁了就去访问数据库，这样就完全可以解决缓存穿透问题

布隆过滤器的应用

比如说输入用户名的时候，可以马上检测出该用户名是否存在，黑名单机制，单词错误检测等