Redis笔记
一、redis入门
redis介绍
是什么？
- 基于内存的K/V存储中间件
- NoSQL键值对数据库
redis不仅是数据库，他还可以做消息队列
关系型SQL和NoSQL的对比
其中应用场景：
Redis特征
- 支持各种数据类型
- 单线程，每个命令的执行具备原子性,中途不会执行其他命令
- 高性能，低延迟
- 支持数据持久化
- 支持主从、分片集群
- 支持多语言客户端
Redis常用命令
Redis的命令根据数据结构分为多个组
1）redis 命令集 http://www.redis.cn/commands.html
2）redis-cli help 命令查看，help[command]可以查看某个具体命令，help@xxx可以查看某个分组下的命令
通用命令
- set key value
- get key
- key pattern 模糊搜索多个Key
- del key…
- exists key 判断key是否存在
- expire key 设置过期时间
- ttl key 查询剩余存活时间，未设置过期时间则为-1
Redis 基本数据类型
String类型
支持存储字符串、数字、浮点数
单key的value最大不能超过512M
实际使用时，通常用冒号连接多个次来拼接key，比如【项目名】:【业务名】:【类名】:【id】
在使用GUI时，会自动根据冒号来划分层级，浏览更方便
Hash类型
值是一个Hash结构(类似于java的HashMap)
常用命令:
List类型
理解为Java的LinkList 双向链表，特点是有序、插入删除快、但查询性能一般：
常见命令如下:
Set类型
集合：类似于java的HashSet，特点是单set内集合不能重复，查找性能高
常见命令如下:
SortedSet类型
有序集合，在set的基础上给每个元素多存了一个分数，类似于value类型为整形的Hashmap
特点：有序、查询性能高、适合于排行榜、统计TopN
常用命令:
Redis 客户端
主流客户端
可以在官网查看所有客户的以及推荐的客户端:https://redis.io/docs/clients
对于Java主要以下几种：
其中jedis的命令和原生redis命令行命令一致，学习成本最低（线程不安全，需要搭配连接池使用），letture和spring兼容型最好（spring Data Redis默认集成），基于netty性能最高，Redisson提供了和java集合一致的分布式集合，适用于更复杂的业务场景
Spring Data Redis
spring Data 封装了一系列数据访问操作，spring Data Redis则是封装了Jedis 、 Lettuce 两个Redis 客户端的操作，提供了统一的Redis Template 来操作Redis。
RedisTemplate 针对不同的Redis数据结构提供了不同的API，划分更明确：
注意：在使用spring Data Redis中使用Lettuce 线程池的话，要额外引入apache commons-pool2 依赖。
RedisTemplate 序列化
RedisTemplate 默认使用JDK原生序列化器，可读性差、内存占用大、因此可以使用以下两种方式改变序列化机制：
- 自定义RedisTemplate，指定key和value的序列化器
- 使用自带的StringRedisTemplate，key和value使用默认的String序列化器，仅支持写入String类型的key和value，因此需要将对象序列化为String来写入redis，从redis读取时也要反序列化。
共享Session（单点登录）
介绍
为什么要共享session？
防止多个后端服务器存储不一致，导致用户访问时出现未登录的情况
如何实现共享Session
使用独立的内存存储来实现Session
实现
key如何设计
为了安全性，随机生成Token，而不是拼接用户信息，防止恶意伪造或爆破
选用何种value数据结构放入用户信息
string还是hash？有两种方案：
1、先在程序中将对象进行Json序列化，再以String类型写入
2、直接以Hash结构写入
因为用户信息时对象，建议选择hash数据结构，占用内存更少、且支持对单个字段的增删改查
具体流程：
注意事项
- 写入Redis的数据一定设置过期时间
- 存入redis的数据尽量保证精简安全，比如存入用户信息时移除密码等敏感数据
- 已登陆的用户访问系统后，记得刷新token过期时效，并且访问任何路径时都要刷新token，而不仅是登录的路径，可以新增一层的拦截器来实现token刷新，如下图：
缓存
什么是缓存？
临时存放数据的缓冲区
缓存无处不在：
为什么需要缓存？
提前准备好数据，便于更快读写
缓存不是银弹，要权衡利弊！
优点：
- 降低后端节点负载
- 提高数据读取性能
缺点：
- 额外引入中间间，增加运维成本
- 额外开发和解决缓存带来的问题，提高开发成本
- 需要保证数据一致性
实现
关键流程
- 暂无缓存，从数据库读取，然后设置缓存
- 已有缓存，直接读取
缓存更新
几种常见的策略，一般选择主动更新+超时剔除兜底：
主动更新缓存的几种方法：
- cache aside ：通过人工编码，在更新数据库的同时更新缓存
- write/read through ：使用现成的数据写入服务，原理就是将缓存和数据库整合为一套服务，让服务来维护数据库和缓存的一致性，我们只需要写数据即可，写完数据后缓存就有，我们不用关心数据一致性问题。
- write behind cache：只更新缓存，通过异步线程定期将缓存的数据持久化到数据库
是选择删除缓存还是更新缓存？
建议选择删除缓存，等待下次查询时自动设置缓存，做到随用随取，可以避免每次更新数据库后都更新缓存的无效操作
多线程情况下，如何保证缓存和数据库的一致性？
- 单机：使用本地事务
分布式，使用TCC等分布式事务
先删除缓存还是先操作数据库
下图为两种方式在多线程下可能存在的问题：
两种方案都不能做到百分百一致性，但是建议选择先操作数据库，再删除缓存，出现问题的概率会更低，可以配合延时双删保证缓存一定被删除
问题及解决
缓存穿透
客户端请求的数据在缓存和数据库中都不存在的，这样缓存永远不会生效，这些请求都会打到数据库。
解决方案：
- 缓存空值：比如向redis中塞一个空字符串，注意可以给空对象设置一个短的过期时间，或者在新增数据时，强制清除下对应的缓存，防止查出来的为null；
- 布隆过滤
初始化布隆过滤器：在Redis中，可以使用BitSet数据结构来实现位数组。可以使用SETBIT命令初始化位数组，将所有位都设置为0。
插入缓存数据：在缓存中存储正常的数据。对于每个要插入的数据，使用多个不同的哈希函数计算出多个哈希值，并将对应的位数组位置设置为1，表示该数据存在。
判断请求是否合法：当有请求到达时，先使用相同的多个哈希函数计算出对应的哈希值，并检查位数组中对应的位置是否都为1。如果有任何一个位置为0，则说明该请求的数据一定不存在，可以直接返回缓存未命中的结果，避免访问数据库。
处理缓存未命中：如果布隆过滤器判断请求的数据可能存在（即所有位置都为1），则继续从缓存中查询数据。如果缓存中也不存在该数据，则返回缓存未命中的结果。
通过使用布隆过滤器，可以在O(1)的时间复杂度内快速判断请求是否合法，减少对数据库的不必要访问，从而提高系统性能和安全性。然而，需要注意的是，布隆过滤器是一个概率型数据结构，存在一定的误判率。因此，在使用布隆过滤器时，需要根据实际情况进行调整，以平衡误判率和空间开销。
预防做法：
- 增加对请求数据的校验 ，比如id>0
- 增加对数据格式的控制，比如id设置为10位，不为10位的请求直接拒绝
- 增加用户权限校验
- 通过限流来保护数据库
缓存雪崩
缓存雪崩是指在同一时间大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力
解决思路：
- 不让key同时失效（可以为key设置时间时添加后缀，使到期时间相差几分钟，防止同一时间大量key失效）
- Redis集群，提高服务可用性，不同Redis之间可以实现数据同步
- 为缓存添加多级缓存（nginx，redis，jvm等，最后查询数据库）
- 给缓存业务添加降级限流策略
- 尽量不让redis宕机
缓存击穿
缓存击穿问题也叫热点key问题，就是一个被高并发访问并且缓存重建
业务较复杂的key失效，无数的请求在瞬间给数据库带来了巨大冲击
两种解决方案：
- 互斥锁：只有一个线程负责缓存重建，其他线程拿不到锁，就等待
- 逻辑过期：key设置为永不过期，在value中记录过期时间，业务中根据这个时间判断缓存是否有效，如果缓存已过期，只有一个线程抢到锁，然后再次判断缓存是否存在，开启独立的线程去更新缓存，然后立即返回过期数据，其他抢不到锁的线程也立即返回过期数据，不用等着释放。
如图所示：
两种方式都使用了互斥锁来降低重建的开销。
方案优缺点对比:
以上几种缓存常见问题的解决方案都是和业务无关的，因此可以封装自己的缓存操作封装类。
互斥锁解决思路：
1 | private TbShop quarywithStuck(Long id) throws InterruptedException { |
逻辑过期解决思路：
![L(3XY$%$`A~)%[MN)Q9GS7.png
具体代码实现：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
public Result queryShopById(Long id) throws InterruptedException {
// 缓存穿透解决方案
// TbShop tbShop = quarywithNull(id);
// 利用互斥锁解决缓存击穿
// TbShop tbShop = quarywithStuck(id);
TbShop tbShop = quarywithLocalExpire(id);
if (Optional.ofNullable(tbShop).isPresent()) {
Result.fail("该店铺不存在");
}
return Result.ok(tbShop);
}
/**
* 上锁
*
* @return
*/
private Boolean tryLock(String key) {
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 50, TimeUnit.SECONDS);
return BooleanUtil.isTrue(aBoolean);
}
private void relaseLock(String key) {
stringRedisTemplate.delete(key);
}
public void setExpre(Long id, Long expireSecond) {
TbShop tbShop = this.getById(id);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
RedisData redisData = new RedisData();
redisData.setData(tbShop);
redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSecond));
stringRedisTemplate.opsForValue().set(CACHE_QUERYSHOP_ID + id, JSONUtil.toJsonStr(redisData));
}
//建立线程池
public static final ExecutorService CHCHE_REBUTLD_EXECUTOR = Executors.newFixedThreadPool(10);
/**
* 逻辑过期解决缓存击穿
*
* @param id
* @return
*/
private TbShop quarywithLocalExpire(Long id) throws InterruptedException {
// 设置key
String Key = CACHE_QUERYSHOP_ID;
// 查缓存
String json = stringRedisTemplate.opsForValue().get(Key + id);
// 未命中 为空，返回null
if (StrUtil.isBlank(json)) {
return null;
}
// 命中，判断是否过期
RedisData redisData = JSONUtil.toBean(json, RedisData.class);
JSONObject data = (JSONObject) redisData.getData();
TbShop tbShop = JSONUtil.toBean(data, TbShop.class);
LocalDateTime expireTime = redisData.getExpireTime();
if (expireTime.isAfter(LocalDateTime.now())) {
// 未过期，直接返回店铺信息
return tbShop;
}
// 已过期，尝试获取互斥锁
Boolean aBoolean = tryLock(LOCK_SHOP_KEY + id);
if (!aBoolean) {
log.info("未获取到锁");
// 获取失败，返回旧数据
return tbShop;
} else {
// 获取成功，开启独立线程，返回旧数据
log.info("已获取到锁");
CHCHE_REBUTLD_EXECUTOR.submit(() -> {
try {
this.setExpre(id, 20L);
} catch (Exception e) {
throw new BusinessException(ErrorCode.SYSTEM_ERROR, e.toString());
} finally {
// 释放锁
relaseLock(LOCK_SHOP_KEY + id);
}
});
}
return tbShop;
}
封装工具类
引入模块:1
2
3
4
5
6
7
8
9
10
11
12
13
14
15 <dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.7.17</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
解决问题（缓存穿透（空值）、缓存击穿(逻辑过期)）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
163package com.xiaozhang.utils;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.xiaozhang.entity.domain.TbShop;
import com.xiaozhang.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import static com.xiaozhang.utils.RedisConstants.*;
import static com.xiaozhang.utils.RedisConstants.LOCK_SHOP_KEY;
/**
* @author:22603
* @Date:2023/5/27 13:59
*/
public class RedisClient {
private final StringRedisTemplate stringRedisTemplate;
public RedisClient(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
public void set(String key, Object value, Long time, TimeUnit unit) {
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, unit);
}
public void setwithLocalExpire(String key, Object value, Long time, TimeUnit unit) {
RedisData redisData = new RedisData();
redisData.setData(value);
redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
}
/**
* 缓存穿透利用空值解决
*
* @param id
* @return
*/
public <R, ID> R quarywithNull(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbfallck, Long time, TimeUnit unit,Long cacheWithNull,TimeUnit unitWithNull) {
// 设置key
String Key = keyPrefix + id;
// 查缓存
String json = stringRedisTemplate.opsForValue().get(Key);
if (StrUtil.isNotBlank(json)) {
R r = JSONUtil.toBean(json, type);
log.info("通过缓存查询数据!");
return r;
}
if (json != null) {
log.info("通过缓存查询数据为空值!");
return null;
}
// 查库
R result = dbfallck.apply(id);
if (Optional.ofNullable(result).isPresent()) {
try {
this.set(Key, result, time, unit);
log.info("通过数据库查询数据并写入缓存!");
return result;
} catch (Exception e) {
throw new BusinessException(ErrorCode.SYSTEM_ERROR, "缓存写入失败!");
}
} else {
// 将空值写入redis
stringRedisTemplate.opsForValue().set(Key, "");
stringRedisTemplate.expire(Key, cacheWithNull, unitWithNull);
log.info("通过数据库查询数据不存在，并写空值到缓存中!");
return null;
}
}
/**
* 上锁
*
* @return
*/
private Boolean tryLock(String key) {
Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 50, TimeUnit.SECONDS);
return BooleanUtil.isTrue(aBoolean);
}
/**
* 解锁
* @param key
*/
private void relaseLock(String key) {
stringRedisTemplate.delete(key);
}
//建立线程池
public static final ExecutorService CHCHE_REBUTLD_EXECUTOR = Executors.newFixedThreadPool(10);
/**
* 逻辑过期解决缓存击穿
*
* @param id
* @return
*/
public <R,ID> R quarywithLocalExpire(String keyPrefix,String lockeyPrefix, ID id, Class<R> type, Function<ID, R> dbfallck, Long time, TimeUnit unit) {
// 设置key
String Key = keyPrefix+id;
// 查缓存
String json = stringRedisTemplate.opsForValue().get(Key);
// 未命中 为空，返回null
if (StrUtil.isBlank(json)) {
return null;
}
// 命中，判断是否过期
RedisData redisData = JSONUtil.toBean(json, RedisData.class);
JSONObject data = (JSONObject) redisData.getData();
R r = JSONUtil.toBean(data, type);
LocalDateTime expireTime = redisData.getExpireTime();
if (expireTime.isAfter(LocalDateTime.now())) {
// 未过期，直接返回店铺信息
return r;
}
// 已过期，尝试获取互斥锁
Boolean aBoolean = tryLock(lockeyPrefix + id);
if (!aBoolean) {
log.info("未获取到锁");
// 获取失败，返回旧数据
return r;
} else {
// 获取成功，开启独立线程，返回旧数据
log.info("已获取到锁");
CHCHE_REBUTLD_EXECUTOR.submit(() -> {
try {
log.info(lockeyPrefix + id);
Thread.sleep(1000);
R result = dbfallck.apply(id);
this.setwithLocalExpire(Key,result,time,unit);
} catch (Exception e) {
throw new BusinessException(ErrorCode.SYSTEM_ERROR, e.toString());
} finally {
// 释放锁
relaseLock(lockeyPrefix + id);
}
});
}
return r;
}
}
分布式全局ID生成
为什么需要分布式全局ID生成器?
1.对于订单这种数据，数据库自增的规律性太明显，会暴露一些信息（比如根据昨日和今日的订单号差值看出销
量)
2．数据量过大时，不同表的id分别自增，容易出现id冲突
分布式全局ID生成应满足的特点:
1．唯一:整个系统每个id都是唯一的
2．递增:虽然不连续，但整体ID保持递增，有利于数据库创建索引(也符合自然规律)
3．安全:不能通过id看出敏感业务信息
4．高可用:作为核心服务，不能挂掉，否则会影响新数据的生成5．高性能:作为频繁调用的服务，性能—定要高
几种常见的ID生成方法，建议根据自己的实际需求选择和设计算法:
- 雪花算法:性能更高，引入机器序号，但依赖全局时钟
- 数据库自增:单独的自增表，所有id全从这个表取。但性能没有Redis高
- UUID:随机生成十六进制字符串，性能高，但是乱序、字符串会占用更多空间
- Redis自增ID:利用incr命令实现单key的自增
Redis自增ID完全可以满足以上几个分布式全局ID的特点。
设计实现
以下是一种ID生成的设计，使用Long类型存储ID![T]6]UP)NBR)~0HR2H68`_%G.png](https://cdn.nlark.com/yuque/0/2023/png/33551426/1685174529136-c1643a46-bf91-43de-baa6-3a59afc22dd8.png#averageHue=%23f6f4f4&clientId=ud5f11d0f-88e8-4&from=paste&height=516&id=u416a5458&originHeight=568&originWidth=1172&originalType=binary&ratio=1.100000023841858&rotation=0&showTitle=false&size=162891&status=done&style=none&taskId=ua1273b1b-761b-4b35-996c-309a0a8cf88&title=&width=1065.4545223614407)
使用Redis的incr命令，可以实现后32位的原子性递增。
Redis的key可以设计为[业务]:[类型]:[日期],这样每天都会从1开始生成序列。如果用单key，可能出现生成的序号数溢出2^32的情况
1 | package com.xiaozhang.config; |
优惠券秒杀流程
秒杀业务的核心流程是:判断日期和库存、扣减库存、创建订单
如下图:
注意，秒杀库存信息和商品信息最好是独立的两张表，不要放在一起影响性能。
订单超卖问题
出现的原因：有多个人同时下单，在库存扣减前大家查库存都大于0，都触发了减库存的操作
解决方法：
1）悲观锁:假定每次并发都会冲突，所以干脆给操作整体加锁，将并发改为同步执行。可以通过synchronized关键字实现。
优点是实现简单，缺点是严重影响性能(大家可以同时抢购)。
2)乐观锁:假定并发不一定会冲突，所以不加锁，而是通过判断数据是否在查出来之后被其他线程修改过，来决定是否允许操作。
乐观锁主要有版本号法和CAS两种实现方式。
版本号法:
给数据增加一个版本号字段，每次修改操作版本号＋1，就可以通过版本号来判断数据是否有被修改。
CAS是对乐观锁的简化，即直接用一个每次都会查询和更新的字段来代替版本号，比如库存stock 字段:
优点:性能好
缺点:存在成功率低的问题(很多人查到的版本号是一样的，结果只能有一个人操作成功)，可以使用分段锁来改进。比如将100个库存分为10份，大家分别抢这10份。
对于以上这种场景，其实不用判断stock是否变化，可以直接判断stock > 0，从而保障成功率。
一人一单
对应视频P54，这集封神
优惠券或者秒杀活动的目的是为了吸引新用户，因此不能让一个用户把所有东西都抢走了。所以需要额外判断用户当前下单数是否> 0。
这一步操作在多线程场景下依然会出现问题:新用户第一次进来同时抢10次，结果判断下单数都是0，然后就都抢成功了。
所以还是需要加锁，因为订单是新创建的数据，所以无法使用乐观锁，使用悲观锁实现。
单机实现
单机部署后端服务器时，可以使用java自带的synchronized关键字作为悲观锁
要注意几个细节问题：
1）synchronized的范围不能太大，不能锁住整个对象，会严重影响性能，因为是一人一单 ，所有可以每个用户独立的一把锁。
注意:锁住 对象时要用toString().intern()，保证同id的用户始终是同一个用户，单单tostring，会new一个String，相当于一个新对象。1
2
3
4
5 synchronized (id.toString().intern()) {
// 获取事务代理对象
TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
2)synchronized必须在使用@Transactional注解的方法外层使用，因为@Transactional是使用代理对象，，在方法执行完才能提交事务，如果把synchronized写在事务方法里，提交事务前释放锁，但此时数据未更新，其他线程依然能获取锁并顺利执行，依然会存在线程问题
3）调用事务方法不能用this对象，因为@Transactional注解实际是调用spring生成的代理对象方法，如果调用this对象，会无法使用事务功能，所以要获取代理对象并调用。
代码实现：
分布式实现
synchronized只对单个jvm有效，多机部署可能同时有多个不同的jvm线程访问已加锁的方法。
具体原理：
单机条件下，只有一个tomcat，一个tomcat服务器相当于有一个全新的jvm，在jvm内部维护了一个锁监视器对象，在Id相同的情况下，只能存放一个线程，其他线程进来时不允许访问，维护的是同一个用户。在集群模式下，会有多个jvm存在，每个jvm内部有自己的锁，每个锁相互独立，多个线程同时运行，会存在线程安全问题。
如图：
因此，我们不能把锁存储在单个服务器上，而是应该使用一个集中的存储来管理锁，所以进程都能读到他，这就需要分布式锁。
分布式锁
分布式锁：满足分布式系统或者集群模式下多个进程可见并且互斥的锁
分布式锁的两个基本特征：
- 多进程（线程）可见（读写）
- 互斥
还具备的特征：
- 高可用：不能挂机，不能经常出现问题
- 高性能（并发）：读写快
- 安全性：不能出现死锁
实现方法
注意有三种：
其中，mysql的实现成本最低、redis性能最高，Zookeeper可以实现，但是不推荐（保证的是强一致性而不是性能和高可用性，cp模式）
Redis 分布式锁实现
获取锁：
- 使用setnx命令设置lock（本质是键值对），保证只有一个线程取锁成功。并执行业务逻辑，其他线程可以重试或返回失败
- 必须setnx指定lock、过期时间（满足安全性）
注意事项：
- 为了防止setnx后宕机导致死锁，必须使用set 【key】 ex nx 的原子命令，保证每个lock都有过期时间
- 锁的key建议设计包含userid，保证多个用户可以并发执行，而不是多个用户抢同一把锁。
释放锁：
- 主动是否：业务执行完删除key，注意释放锁的逻辑放在finlly中
- 超时自动释放（key过期）
流程：
误删情况
情况1
如果线程A执行业务时间过长，锁提前过期了，另一个线程B拿到锁并执行业务流程，这是A突然又执行完了，结果误删了线程B加的锁，会导致新的线程C又可以拿到锁，从而又出现线程安全问题
情况1解决方案
在获取锁的时候value中存入【本机标识+当前线程id】，释放锁时检测value必须等于该值，是自己的锁才可以释放。
注意：不能只在value中存入线程id，因为多个机器的线程id可能是一样的，仍然可能出现问题，因此可以给每个机器生成一个唯一标识，再拼接id
流程如下：
情况2
假设线程A释放锁时判断了是自己的锁，但这时jvm触发了stop the world，线程A卡住了，然后锁超时释放了，线程B拿到了锁的执行业务，这时，线程A醒来了，删除了锁key，线程c正好进来啦，拿到了锁并执行，又出现了线程安全问题。
情况2 - 解决方案
问题的本质是判断锁value和删除锁是两个动作，不具备原子性！
可以使用redis lua脚本，将多个redis 命令放到一个脚本中，整个脚本的执行具有原子性。
Lua语言是轻量级脚本语言，很方便嵌入各种应用程序中。
https://www.runoob.com/lua/lua-tutorial.html
Lua脚本可以使用redis.call 调用redis命令，并支持传递动态参数：
用Lua脚本释放锁脚本如下：
Redis分布式锁的实现思路
- 利用set nx ex 获取锁，nx确保互斥，只有一个线程能拿到锁，ex兜底，过期锁自动删除
- 释放锁时先判断线程标识是否和自己一致，一致则删除。
特性：
- 不可重入:同一个线程无法获取同一把锁（递归调用或调用的子函数抢同一把锁时就会出现死锁)
- 不可重试:没抢到锁就失败了
- 超时释放:业务未执行完，锁就超时释放了
- 主从一致性:主节点设置锁成功，还未及时同步到从节点，这时主节点宕机，从节点被选为主节点。但此时从
节点还没有锁，仍可以抢锁成功。
要自己解决这些问题，非常麻烦，所以我们一般会选择现成的类库，比如Redisson。
Redisson
Redisson 是java的Redis高级客户端，提供了各种现成的分布式工具类方便我们使用Redis
中文文档：https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
1）引入独立的Redisson包1
2
3
4
5<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.5.0</version>
</dependency>
2)创建一个Redisson客户端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
37package com.xiaozhang.config;
import lombok.Data;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Redisson 配置
* @author 22603
*/
public class RedissonConfig {
private String host;
private String password;
private String port;
public RedissonClient redissonClient() {
// 1. 创建配置
Config config = new Config();
String redisAddress = String.format("redis://%s:%s", host, port);
config.useSingleServer().setAddress(redisAddress).setPassword(password).setDatabase(0);
// 2. 创建实例
RedissonClient redisson = Redisson.create(config);
return redisson;
}
}
3)使用Redisson的lock1
2
3
4
5
6
7
8
9
10
11RLock lock = redissonClient.getLock("lock:order:" + id);
boolean tryLock = lock.tryLock();
if (!tryLock) {
return Result.fail("不允许同一用户重复下单!");
}
try {
TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
} finally {
lock.unlock();
}
如何实现可重入锁？
目的：保证同一个线程可以多次获取同一把锁
解决思路：在锁的value中额外保存当前线程获取锁的次数，每次获取锁+1，释放锁-1，当次数为0时真正可以了删除key
Redisson 底层也是通过lua脚本实现可重入锁
采用hash结构来存储锁信息：
流程：
注意：
- 所有的判断和操作都需要使用Lua脚本来保证原子性
- 每次获取和释放锁时要重置锁的有效期，就像抢到新锁一样，给业务充分的执行时间
获取锁的Lua脚本:1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22--获取锁的key
local key=key[1];
--获取线程唯一标识
local thread =ARGV[1];
--所得自动释放时间
local releaseTime= ARGV[2];
--判断锁是否存在
if (redis.call('exists',key)==0) then
--不存在，获取锁
redis.call('hset',key,thread,'1');
--设置有效期
redis.call('expire',key,releaseTime)
end
--若锁存在 判断锁是否为自己的
if (redis.call('hexists',key,thread)==1) then
--是自己的，则统计数+1
redis.call('hincrby',key,thread,'1');
--设置有效期
redis.call('expire',key,releaseTime);
return 1; --返回结果
end
return 0;
释放锁的Lua脚本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---释放
--- Generated by Luanalysis
--- Created by 22603.
--- DateTime: 2023/6/10 9:34
---
---获取key
local key = Key[1];
--获取唯一标识
local thread = ARGV[1];
--获取锁的自动释放时间
local releaseTime = ARGV[2];
--判断锁是否为自己的锁
if (redis.call('HEXISTS', key, thread) == 0) then
return nil;
end
--是自己的锁，则冲入次数-1
local count = redis.call('HINCRBY', key, thread, -1);
--如果
if (count > 0) then
redis.call('EXPIRE', Key, releaseTime);
return nil;
else
redis.call('DEL', key);
return nil;
end
如何实现重试获取锁？
基于redis pub/sub发布订阅机制，如果获取锁失败，则阻塞订阅释放所得消息，当锁被释放时，会触发推送（告诉其他线程我释放锁），然后其他线程再重试获取，如此以往，直到超时。
如何防止锁提前释放？
基于看门狗机制，如果不手动设置释放时间，默认设置30s过期，并且给当前锁注册一个定时任务，该定时任务每隔1/3的锁释放时间（一般是10s），会重置锁的过期时间（递归调用，一次续期完了再）。
需要考虑两个问题：
- 如何保证同一个锁只注册一个定时任务？
- 如何防止无线续期？
要解决这些问题，使用全局ConcurrentHashMap来管理锁=>任务信息，key为锁的id，从而保证唯一，当某个锁释放时，从全局ConcurrentHashMap中取处定时任务并取消，然后把锁的消息从Map中删除即可。
最终，完整的分布式锁流程如下：
Redisson主从一致性问题
如果使用主从复制的多台redis集群，可能出现主从节点设置的锁状态不一致的问题。
可以使用Redisson的MultiLock（联锁）来解决，核心思想是开启多个独立的Redis主节点，设置锁时必须在所有的主节点都写入成功，才算设置成功。
这样做哪怕有部分节点挂掉，其他线程也无法setnx全部成功，就不会出现重复执行业务的问题。
如图：
实现MultiLock的几个关键：
- 遍历所有节点，依次设置锁，并使用列表记录所有主节点的锁是否设置成功。
- 只要有一个节点设置不成功，就要释放锁，从头来过。
- 因为不同节点设置成功的时间不同，所以所有锁设置成功后，要统一设置过期时间（但如果leaseTime=-1就不用设置，因为开启了看门狗机制会自动续期）
- 释放锁时间（leasetime）必须要大于抢锁最大等待时间（waitTime），否则可能出现第一个节点抢到锁，最后一个节点还没抢到锁，之前的锁已经超时释放了，所有如果指定了waitTime和leaseTime，默认leasetime=waitTime*2
MultiLock最安全，但同样会带来很大的运维成本。
秒杀业务优化
优化思路：
- 串行改并行：原本由一个线程的操作改为由两个或多个线程同时执行，比如一个线程负责判断秒杀资格，一个线程负责减库存+创建订单（写操作）
- 同步改异步：判断完秒杀资格后，就可以返回订单id给前端，其余的写操作可以异步执行。
- 提高判断秒杀资格的性能：读DB改为读redis
优化后的流程如下：
1）将库存信息提前缓存到redis中，并使用set来记录用户是否下单，实现仅在redis里判断秒杀资格
2）将判断秒杀资格的逻辑封装为Lua脚本，保证原子性，原业务流程调用即可、
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23-- 优惠卷id
local voucherId = ARGV[1]
-- 用户id
local userId = ARGV[2]
-- 缓存key
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
-- 1、判断库存是否充足
if(tonumber(redis.call('get', stockKey)) <= 0) then
-- 1.1 不充足直接返回1
return 1
end
-- 1.2 充足
-- 1.2.1 判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 买过返回2
return 2
end
-- 未买过返回 扣库存
redis.call('incrby',stockKey,-1);
--下单
redis.call('sadd',orderKey,userId);
return 0
3）确认有秒杀资格后，将订单信息传递给阻塞队列，单个独立线程串行从队列中取出信息并异步下单
这样一来，可以大大提高性能。
阻塞队列可以使用jdk原生的BlockingQueue实现，记得指定队列容量。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
private StringRedisTemplate stringRedisTemplate;
private TbSeckillVoucherService tbSeckillVoucherService;
private RedisIdWorker redisIdWorker;
// 创建阻塞队列
private BlockingQueue<TbVoucherOrder> blockingDeque = new ArrayBlockingQueue<>(1024 * 1024);
// 提供了一系列方法来管理线程池和执行任务
private static final ExecutorService SECKILL_ORDER_SERVICE = Executors.newSingleThreadExecutor();
//类加载完成后执行初始化任务
private void init() {
SECKILL_ORDER_SERVICE.submit(new VoucherOrderHander());
}
//初始化代理对象
private TbVoucherOrderService proxy;
// 开启一个独立的线程
private class VoucherOrderHander implements Runnable {
public void run() {
while (true) {
try {
TbVoucherOrder voucherOrder = blockingDeque.take();
handleVoucherOrder(voucherOrder);
} catch (InterruptedException e) {
log.info("异步抢购失败:" + e);
}
}
}
}
private RedissonClient redissonClient;
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}
/**
* 优惠卷秒杀优化
* 串行改并行：原本由一个线程的操作改为由两个或多个线程同时执行
* 比如一个线程负责判断秒杀资格，一个线程负责减库存+创建订单（写操作）
* 同步改异步：判断完秒杀资格后，就可以返回订单id给前端，其余的写操作可以异步执行。
* 提高判断秒杀资格的性能：读DB改为读redis
*
* @param voucherId
* @return
* @throws InterruptedException
*/
public Result seckillVoucher(Long voucherId) throws InterruptedException {
Long userId = UserHolder.getUser().getId();
// 1、执行lua脚本
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString()
);
// 2、判断结果为0
int value = result.intValue();
if (value != 0) {
// 2.1 不为0，代表没有购买资格
return Result.fail(value == 1 ? "库存不足" : "不能重复下单");
}
// 2.2 为0 ，有资格，把下单信息保存到阻塞队列
// 2. 保存到阻塞队列
// 获取代理对象
proxy = (TbVoucherOrderService) AopContext.currentProxy();
TbVoucherOrder tbVoucherOrder = new TbVoucherOrder();
long order = redisIdWorker.nextId("order");
tbVoucherOrder.setId(order);
tbVoucherOrder.setVoucherId(voucherId);
tbVoucherOrder.setUserId(UserHolder.getUser().getId());
// 将订单添加到阻塞队列中
blockingDeque.add(tbVoucherOrder);
return Result.ok(order);
}
private void handleVoucherOrder(TbVoucherOrder tbVoucherOrder) throws InterruptedException {
// 1.获取对象
Long userId = tbVoucherOrder.getUserId();
// 2.获取锁对象
RLock lock = redissonClient.getLock("lock:order:" + userId);
// 3.获取锁
boolean tryLock = lock.tryLock(1L, TimeUnit.MINUTES);
// 4.失败返回
if (!tryLock) {
return;
}
// 5.代理对象
try {
proxy.createVoucherOrder2(tbVoucherOrder);
return;
} finally {
lock.unlock();
}
}
public void createVoucherOrder2(TbVoucherOrder voucherOrder) {
// 获取订单数量
Long id = voucherOrder.getId();
int count = query().eq("user_id", id).eq("voucher_id", voucherOrder.getVoucherId()).count();
if (count > 0) {
log.info("重复购买");
return;
}
// 扣减
boolean result = tbSeckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherOrder.getVoucherId()).ge("voucher_id", 0).update();
if (!result) {
log.info("库存不足");
return;
}
if (!this.save(voucherOrder)) {
return;
}
}
优惠卷秒杀
原始方法
流程图：
代码逻辑：
- 判断秒杀是否开始
- 判断库存
- 单机情况下为解决超卖问题，为确保原子性，将获取订单数量、判断是否购买、扣减、保存订单信息等和数据库相关的写操作加悲观锁
- 分布式情况下，为实现带来的数据不一致性问题，采用分布式锁实现
代码实现：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
public Result seckillVoucher(Long voucherId) throws InterruptedException {
TbSeckillVoucher seckillVoucher = tbSeckillVoucherService.getById(voucherId);
if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
return Result.fail("秒杀未开始");
}
if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
return Result.fail("秒杀已结束");
}
if (seckillVoucher.getStock() < 1) {
return Result.fail("库存不足");
}
Long id = UserHolder.getUser().getId();
// ① synchronized 单机实现 锁的整个方法
// synchronized (id.toString().intern()) {
// 获取事务代理对象
// TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
// return proxy.createVoucherOrder1(voucherId);
// }
// ② 分布式实现 锁的整个方法
// SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + id, stringRedisTemplate);
RLock lock = redissonClient.getLock("lock:order:" + id);
boolean tryLock = lock.tryLock(1L, TimeUnit.MINUTES);
if (!tryLock) {
return Result.fail("不允许同一用户重复下单!");
}
try {
TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder1(voucherId);
} finally {
lock.unlock();
}
}
/**
* 分布式解决优惠卷秒杀
*
* @param voucherId
* @return
*/
public Result createVoucherOrder1(Long voucherId) {
// 获取订单数量
Long id = UserHolder.getUser().getId();
int count = query().eq("user_id", id).eq("voucher_id", voucherId).count();
if (count > 0) {
return Result.fail("该用户已购买一次!");
}
// 扣减
boolean result = tbSeckillVoucherService.update().setSql("stock=stock-1").eq("voucher_id", voucherId).ge("voucher_id", 0).update();
if (!result) {
return Result.fail("失败!");
}
TbVoucherOrder tbVoucherOrder = new TbVoucherOrder();
long order = redisIdWorker.nextId("order");
tbVoucherOrder.setId(order);
tbVoucherOrder.setVoucherId(voucherId);
tbVoucherOrder.setUserId(UserHolder.getUser().getId());
if (!this.save(tbVoucherOrder)) {
return Result.fail("失败!");
}
return Result.ok(order);
}
优化思路：
- 将秒杀功能中对优惠卷信息 、订单信息缓存到redis中，为了数据的原子性，通过使用Lua脚本，来进行对redis中优惠卷以及订单信息的判断，最终，若用户首次购买，则将优惠卷信息添加到阻塞队列中，然后返回订单id。
异步开启一个独立的线程，去读取阻塞队列中的优惠卷信息，然后进行减库存以及保存订单信息等对数据库的写操作。
消息队列
jdk阻塞队列可能存在哪些问题？
服务器宕机，内存队列的订单信息全部丢失
- 线程处理错误，已读出单个订单信息，但是没有入库
- 受单个jvm内存限制
所有，我们需要一个独立的队列来存管订单信息，也就是消息队列。
什么是消息队列？
存管消息的队列，也是一种开发常用的中间件。
消息队列包括三个角色：
- 消息队列：用于存放消息，类似于快递柜
- 生产者：发送消息到队列，类似快递员
- 消费者：从队列取走消息，类似于取快递的人
使用消息队列的优点：
- 可以确保消息的安全、不丢失（快递柜上锁）
- 可以解藕生产者和消费者（不用立刻去取快递）
- 独立组件，不影响jvm
- 可以保证消息一定被接受，避免线程处理错误后订单丢失的问题
- 消息是有序的
实现方式
Redis List实现
使用Redis List 的结构作为消息队列，使用LPush模拟生产者发送消息入队，使用BRPOP模拟消费者取出消息，没有消息时会保持阻塞状态，从而实现类似jvm阻塞队列的效果
可以满足消息队列的安全性（redis的持久化机制）、有序、独立内存的要求。
缺点：
- 只能存在单消费者
- 消息获取之后就删除了，无法保证业务一定办理成功
Redis Pub/Sub
使用Redis的订阅发布模式，生产者可以将消息推送给某个channel，多个消费者可以订阅该频道，从而同时得到消息。
用的命令是：
- Publish 推送
- SubScribe 订阅
- PSubScribe 订阅某个表达式匹配的多个频道
这样就可以实现多生产、多消费。也很灵活，一条消息可以发送给一个消费者，也可以发送给多个消费者。
缺点：
- Pub/Sub是一次性的，Redis不会保存发过的消息，没人收消息就会丢失。
- 因为上一点，无法做到持久化
- 客户端虽然可能会缓存收到的消息，但是也有上限，可能还是会丢失消息
不建议使用
Redis Stream
redis 5.0 新推出的数据结构，可以实现单向的消息队列。
核心命令：
- XADD：添加消息/创建队列，消息会自动持久化，不用担心丢失，每隔消息都有唯一的全局唯一id
- XRead：消息可回溯，支持多消费者读，可以指定消息id开始读，支持阻塞读最新消息，有漏读风险。
![4_QC8`J7$%BO_57N$%CO)]8.png](https://cdn.nlark.com/yuque/0/2023/png/33551426/1686735365106-bf4c07a6-f9ba-45c1-a20d-f9c99fb09f5a.png#averageHue=%23122a3d&clientId=u1469bab5-7119-4&from=paste&height=192&id=uffd1de90&originHeight=211&originWidth=1024&originalType=binary&ratio=1.100000023841858&rotation=0&showTitle=false&size=234565&status=done&style=none&taskId=u6408a1e3-4094-42cc-a3b3-15e57f9b323&title=&width=930.9090707321802)
只有这两个命令不够，因为目前只支持阻塞读最新消息，假设处理消息过程又来几条消息，可能出现漏读情况。
为解决上述问题，可以使用Stream的以下特性：
- 消费组：同组的多个消费者可以竞争消费，从而提升消费能力。对应命令为XGROUP、XREADGROUP
- 消费标识：自动记录消费进度，支持从上次未消费的地方开始接着消费，保证每个消息都能按顺序消费
- 消息确认机制：默认消费的消息未pending状态，会放到每个消费者的pending list中，只有消息由消费者确认(ACK)，才会从pending list移除，这样如果消费业务处理异常，可以从pending list开头依次读取未确认的消息，重试处理。（也需要避免无线重试，实在处理不成功，就强制ACK+业务记录日志）
整个消费流程：
在java中操作Redis Strean 两种方法：
- 调用Lua脚本
- 使用Redis Template 的 opsForStrean（）
方案对比：
注意，redis 虽然可以实现较为完备的消息队列，但还是不如专业的消息队列Kafka，RabbitMQ，RocketMQ专业，比如持久化能力差，不支持生产确认，顺序消费性等。
抢购优惠卷业务终极版
redis创建消息队列
1
2
3
4
5
6
7
8// 连接redis服务器
redis-cli
// 认证
redis-cli -h <124.221.169.181> -p <3306>
// 创建消息队列
XGROUP CREATE Stream.orders orders 0 MKSTREAM
// 查看消息队列信息
XINFO STREAM Stream.orderslua脚本，将判断库存，判断用户下单等操作封装到lua脚本中，如果用户没有下单就在redis中扣库存，把订单消息存储到redis中，同时将这个消息发送到队列中，队列采用的是Stream模式，然后返回
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-- 1.参数列表
-- 1.1.优惠券id
local voucherId = ARGV[1]
-- 1.2.用户id
local userId = ARGV[2]
-- 1.3.订单id
local orderId = ARGV[3]
-- 2.数据key
-- 2.1.库存key
local stockKey = 'seckill:stock:' .. voucherId
local orderKey = 'seckill:order:' .. voucherId
-- 2.2.订单key6
-- 3.脚本业务
-- 3.1.判断库存是否充足 get stockKey
if(tonumber(redis.call('get', stockKey)) <= 0) then
-- 3.2.库存不足，返回1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在，说明是重复下单，返回2
return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0在java逻辑代码中，首先进入seckillVoucher接口，然后先执行lua脚本，对于返回结果判断，无购买资格则返回，有购买资格则获取代理对象后返回，通过在类加载完成后初始化的异步线程中，获取消息失败则循环监听消息队列，获取消息队列中的消息后，解析其中的订单信息，然后进行下单，下单成功后返回通过ack消息确认机制进行确认后，将订单消息从队列中删除，抢购失败，则打印日志并获取消息队列中的失败的消息，再次进行消费。
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
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}
// 创建阻塞队列
private BlockingQueue<TbVoucherOrder> blockingDeque = new ArrayBlockingQueue<>(1024 * 1024);
String queueName = "stream.orders";
//初始化代理对象
private TbVoucherOrderService proxy;
// 提供了一系列方法来管理线程池和执行任务
private static final ExecutorService SECKILL_ORDER_SERVICE = Executors.newSingleThreadExecutor();
class VoucherOrderHander implements Runnable {
public void run() {
while (true) {
try {
// 获取消息队列中的消息
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream()
.read(Consumer.from("g1", "c1"), StreamReadOptions.empty()
.count(1).block(Duration.ofSeconds(2)), StreamOffset.create(queueName, ReadOffset.lastConsumed()));
// 判断是否获取成功
if (list == null || list.isEmpty()) {
continue;
}
// 解析消息 中的订单信息
Map<Object, Object> value = list.get(0).getValue();
TbVoucherOrder tbVoucherOrder = BeanUtil.fillBeanWithMap(value, new TbVoucherOrder(), true);
// 如果成功 可以下单
handleVoucherOrder(tbVoucherOrder);
// ack确认
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", list.get(0).getId());
} catch (InterruptedException e) {
log.info("异步抢购失败:" + e);
handlePendingList();
}
}
}
public void handlePendingList() {
while (true) {
try {
// 获取消息队列中的消息
List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream()
.read(Consumer.from("g1", "c1"), StreamReadOptions.empty()
.count(1).block(Duration.ofSeconds(2)), StreamOffset.create(queueName, ReadOffset.from("0")));
// 判断是否获取成功
if (list == null || list.isEmpty()) {
// 失败说明pendinglist中没有消息，直接结束
break;
}
// 解析消息 中的订单信息
Map<Object, Object> value = list.get(0).getValue();
TbVoucherOrder tbVoucherOrder = BeanUtil.fillBeanWithMap(value, new TbVoucherOrder(), true);
// 如果成功 可以下单
handleVoucherOrder(tbVoucherOrder);
// ack确认
stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", list.get(0).getId());
} catch (InterruptedException e) {
log.info("异步抢购失败:" + e);
try {
Thread.sleep(20);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
}
}
/**
* 优惠卷秒杀优化
* 串行改并行：原本由一个线程的操作改为由两个或多个线程同时执行
* 比如一个线程负责判断秒杀资格，一个线程负责减库存+创建订单（写操作）
* 同步改异步：判断完秒杀资格后，就可以返回订单id给前端，其余的写操作可以异步执行。
* 提高判断秒杀资格的性能：读DB改为读redis
*
* @return
* @throws InterruptedException
*/
public Result seckillVoucher(Long voucherId) throws InterruptedException {
Long userId = UserHolder.getUser().getId();
long order = redisIdWorker.nextId("order");
// 1、执行lua脚本
Long result = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(order));
// 2、判断结果为0
int value = result.intValue();
if (value != 0) {
// 2.1 不为0，代表没有购买资格
return Result.fail(value == 1 ? "库存不足" : "不能重复下单");
}
// 2.2 为0 ，有资格，把下单信息保存到阻塞队列
// 2. 保存到阻塞队列
// 获取代理对象
proxy = (TbVoucherOrderService) AopContext.currentProxy();
return Result.ok(order);
}
//类加载完成后执行初始化任务
private void init() {
SECKILL_ORDER_SERVICE.submit(new VoucherOrderHander());
}
点赞
流程：
- 判断用户是否已经点赞
- 更新DB点赞和点赞关系
如果全部数据库操作，性能不高，可以把判断是否已经点赞、维护点赞关系的逻辑移至redis，只用更新数据库点赞即可
可以使用redis set结构，比如key=like:[帖子id],value为去重的已点赞用户id列表，就可以判断是否已经点赞了。
点赞功能的实现可能存在同时点赞的情况，可以参考秒杀优化的逻辑，使用分布式锁、或者lua脚本解决、操作DB的部分可以使用异步、队列之类的机制。
使用分布式锁解决：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
public Result likeBlog(Long id) {
// 1.获取登录用户
Long userId = UserHolder.getUser().getId();
// 2.判断当前登录用户是否已经点赞
String key = BLOG_LIKED_KEY + id;
RLock lock = redissonClient.getLock("lock:like:" + userId);
try {
if (lock.tryLock(0, -1, TimeUnit.MICROSECONDS)) {
Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
if (score == null) {
// 3.如果未点赞，可以点赞
// 3.1.数据库点赞数 + 1
boolean isSuccess = update().setSql("liked = liked + 1").eq("id", id).update();
// 3.2.保存用户到Redis的set集合 zadd key value score
if (isSuccess) {
stringRedisTemplate.opsForZSet().add(key, userId.toString(), System.currentTimeMillis());
}
} else {
// 4.如果已点赞，取消点赞
// 4.1.数据库点赞数 -1
boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update();
// 4.2.把用户从Redis的set集合移除
if (isSuccess) {
stringRedisTemplate.opsForZSet().remove(key, userId.toString());
}
}
return Result.ok();
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
return Result.fail("点赞功能繁忙,请稍后重试!");
}
点赞排行榜
示例需求:展示最先给某个帖子点赞的TopN用户
实现:使用Redis SortedSet(自动按score排序的数据结构)，把用户点赞时间作为score即可，使用ZRange范围查询TopN。
点赞排行榜存在的坑：
- 将点赞用户按时间戳插入redis的zset中时，存在sorce分数高的排在上面，分数低的自动往后移，这里查找了网上关于zset的教程，查询排名时以升序排序（ score 越大排名越靠后），这里是因为quickredis自动排序了（踩了很久的坑），
- 从redis的Zset集合中拿到返回值后，原本的顺序就是点赞的顺序，但是在mapper文件中使用in以及 foreach标签时，会自动对集合中的数据进行排序后，再进行查询，就会导致排行榜点赞顺序与我们的预期不符，在mapper中写sql时通过order by find_in_set(id, #{idStr})实现
1 |
|
1 | <select id="getTopfiveByIds" resultType="com.xiaozhang.entity.domain.TbUser"> |
共同关注
需求:计算两个人共同关注的所有用户思路:本质就是求两个用户关注集合的交集
实现:使用Redis Set数据结构，每个key保存1个用户的关注集合(取关时从集合中移除对应用户)，最后对2个用户对应的set进行intersect取交集操作即可
Feed流关注推送
up主发布新内容，系统给粉丝发送消息
何为feed流？
feed流是指无线下拉刷新的内容列表，给用户沉浸式的体验
两种模式：
- Timeline 时间线模式：所有内容按时间顺序展示，比如朋友圈
- 智能排序模式：根据一定算法推荐用户感兴趣的内容，比如现在的视频网站
解决问题：如何将关注用户的新内容展示到Feed流中？
实现方式
1）拉模式：用户上线打开Feed流时，再获取up推送的内容（主动拉取，比较耗时）
2）推模式：up主发布内容时，直接把内容推送到所有粉丝内容列表里，（粉丝多的大up主写入时很占资源）
3）推拉结合
推模式：粉丝少的up、推送给活跃用户
拉模式：粉丝多的up&&推给非活跃用户
各种方式的优缺点：
一般系统的用户量级小的时候采用推模式，系统用户量大的时候采用推拉结合模式。
推模式Redis实现
因为推送是按照时间排序的，所有可以使用sortedSet数据结构来保存，每一个key表示一个用户的收件箱，value为推送id，sorce为推送时间戳。
分页问题：
每条消息都会动态变化，导致数据重复查询，如图：
因此，可以使用滚动分页，记录每次当前时间戳的下一条开始查询，如图：
可以使用sortedset的zrevrangeBysorce 命令实现，每次查询指定时间戳范围（0，当前最大时间戳）的指定条数的数据。示例命令：
ZRevRangeByScore key 0 maxTimeStamp offset limit。
注意有个问题，可能出现相同时间戳的数据，因此第一次查询时 offset 为 0（从第一条开始查），之后每次查询，offset 为上一页中时间戳最小值的数量，保证不查出重复数据。
比如 score 列表为：5, 4, 4, 3, 2, 1。每页 3 条。
第一次查询 ZRevRangeByScore key 0 9999999 0 3 查出 5、4、4，最小值 4 重复 2 次，即下一次的最大值为 4、offset 为 2。
所以第二次查询为 ZRevRangeByScore key 0 4 2 3 查出 3、2、1
GEO 地理坐标计算
P88 - P90
Redis 支持 GEO 数据结构，可用于存储多个地理坐标、计算坐标间的距离，计算某坐标半径范围内的其他点和距离（实现附近的人）。
GEO 底层是 SortedSet 类型
支持以下命令：
查附近
需求：查询你附近所有的小姐姐，以及每个小姐姐和你的距离，并且按照距离由近到远排序
实现：
1使用 GEOADD 添加所有小姐姐的 id 和位置到一个 key 中
2使用 GEOSEARCH 查询以你当前坐标为中心、指定距离内的所有小姐姐，默认返回的就是由近及远的 SortedSet（value 为 id、score 为和你的距离）
小技巧：
如果我们要根据分类来计算同组内的距离及排名，可以每个类别一个独立的 key，如图：
BitMap
多个二进制位组成的数据结构，每个二进制位对应一个业务的两种互斥状态。比如：10010011。
BitMap 的优点是可以用最小的代码（1 个 bit）来保存业务含义，从而大幅节省空间、实现高效运算等。比如可以把某个二进制位从 0 改为 1。
底层是 string 类型，以字节为单位存储，所以 bit 的长度为 8 的倍数
命令：
签到功能
这是 BitMap 的一个经典应用场景。
如果使用传统数据库来记录用户某日签到状态，每次签到对应 1 行数据，那么将占用几百 byte 的空间。用户量大了后，占用的内存不可估量。
而使用 BitMap，可以用每一个二进制位表示当天的签到状态，0 表示用户已签到、1 表示用户未签到，如下图：
只用 31 bit，就能表示某用户 1 个月的签到状态。
其原理就类似于签到卡，用户可以在一张签到卡上同时对多个不同的日期进行签到，从而避免了重复的用户签名、重复去写日期。
基本实现
用 BITSET 设置某一天对应位置的 bit 值即可，需要查询签到状态时可以用 BITFIELD GET 取出整个 BitMap 的十进制值，再用 位运算 将十进制值转为业务需要的格式。
统计连续签到天数
可以先用 BITFIELD GET 取出整个 BitMap 的十进制值。然后循环右移，依次用 1 去和 BitMap 的最后一位进行 与运算，即可得知当前位数是否为 1（已签到），配合计数器完成连续 1 的统计。
示例代码如下：
注意，循环右移一定要用 >>>（无符号右移，高位补 0），否则可能会死循环。
HyperLogLog
HyperLogLog（HLL）是 Redis 的高级数据结构，使用概率学算法实现，可以用极低的内存实现不重复元素个数的统计。
用法和集合类似，插入元素、求不重复元素个数、合并多个 HLL 等：
优点：占用内存极低，不会超过 16 Kb
缺点：基于概率统计，存在 < 0.81% 的误差
因此，它很适合用于 UV、PV 等数据量大、精度要求不高的统计。
UV / PV 统计
只需要把用户 id 用 PFADD 添加到 HLL 中，用 PFCOUNT 求值即可；还可以用 PFMERGE 合并多天的统计，来估算整月的 UV。
