1 全局ID生成器
一些情境下,使用数据库的ID自增将会产生一些问题。
- 一方面,自增ID规律性明显,可能被猜测出来并产生一些漏洞
- 另一方面,当数据量很大很大很大时,单表数据量可能会受到限制,需要分表,多个表之间的ID自增策略受限
@Component
public class RedisIdWorker {
/**
* 开始时间戳
*/
private static final long BEGIN_TIMESTAMP = 1640995200L;
/**
* 序列号的位数
*/
private static final int COUNT_BITS = 32;
private StringRedisTemplate stringRedisTemplate;
public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
/**
* 根据KeyPrefix生成Id,key为 "icr:" + keyPrefix + ":" + date,每天一个Key,方便统计订单量
* @param keyPrefix
* @return
*/
public long nextId(String keyPrefix) {
// 1、生成时间戳
LocalDateTime now = LocalDateTime.now();
// 转换成当前的秒数
long second = now.toEpochSecond(ZoneOffset.UTC);
long timeStamp = second - BEGIN_TIMESTAMP;
// 2、构造存入的key,并增加count值
String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
Long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
// 3、拼接并返回
return timeStamp << COUNT_BITS | count;
}
}
测试:
// 建一个线程池
private ExecutorService es = Executors.newFixedThreadPool(500);
@Test
void testIdWorker() throws InterruptedException {
// 如果没有CountDownLatch ,
// 由于程序是异步的,当异步程序没有执行完时,主线程就已经执行完了,
// 我们期望的是分线程全部走完之后,主线程再走,所以我们此时需要使用到CountDownLatch
CountDownLatch latch = new CountDownLatch(300);
// 所以使用await可以让main线程阻塞,什么时候main线程不再阻塞呢?
// 当CountDownLatch内部维护的变量变为0时,就不再阻塞,直接放行,
// 调用一次countDown ,内部变量就减少1,我们让分线程和变量绑定, 执行完一个分线程就减少一个变量,
// 当分线程全部走完,CountDownLatch 维护的变量就是0,此时await就不再阻塞
Runnable task = () -> {
for (int i = 0; i < 100; i++) {
long id = redisIdWorker.nextId("order");
System.out.println("id = " + id);
}
latch.countDown();
};
long begin = System.currentTimeMillis();
for (int i = 0; i < 300; i++) {
es.submit(task);
}
latch.await();
long end = System.currentTimeMillis();
System.out.println("time = " + (end - begin));
}
- Runnable接口是一个函数式接口,即只有一个方法。可以通过Lambda函数指定其run方法对应的代码。
2 秒杀下单
- Q1:是否在抢购时间内
- Q2:库存是否充足
- Q3:多个用户并发访问同一张优惠券,需要加锁
乐观锁
在更新数据时去判断有没有其他线程对数据进行了修改。
版本号法:设置一个版本号,每次操作数据会对版本号+1,再提交回数据时,会去校验是否比之前的版本大1 ,如果大1,意味着操作过程中没有人对他进行过修改,则进行操作成功。
CAS法(compare and set):直接使用Stock进行判断,检查修改时库存是否大于0。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mjCxJfYb-1688908161477)(【Redis】秒杀业务设计与分析/image-20230708232543952.png)]
@Override
@Transactional(rollbackFor = Exception.class)
public Result seckillVoucher(Long voucherId) {
// 1、查询优惠券信息
SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
LocalDateTime now = LocalDateTime.now();
LocalDateTime begin = seckillVoucher.getBeginTime();
LocalDateTime end = seckillVoucher.getEndTime();
if (now.isBefore(begin)) {
return Result.fail("秒杀尚未开始!");
}
if (now.isAfter(end)) {
return Result.fail("秒杀已经结束!");
}
int stock = seckillVoucher.getStock();
if (stock <= 0) {
return Result.fail("库存不足");
}
// 2、购买优惠券,加入Order表,Stock更新
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId).gt("stock", 0).update();
if (!success) {
return Result.fail("库存不足");
}
VoucherOrder voucherOrder = new VoucherOrder();
Long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
voucherOrder.setUserId(UserHolder.getUser().getId());
voucherOrder.setVoucherId(voucherId);
save(voucherOrder);
return Result.ok(orderId);
}
一人一单:悲观锁
- 当同一个用户同时向数据库发送多条相同请求时,由于多个请求查找到数据库的结果相同,多个请求均有可能满足条件进行购买,从而产生错误。
- 需要对同一个
user
的购买操作加锁。
将购买逻辑(是否购买过,更新stock,加入order表)封装为一个事务,必须把查询订单信息放在这个函数里,而不是外面。如果先在外面判断是否购买过优惠券,再放入该函数,相当于没有加上锁:
@Override
@Transactional(rollbackFor = Exception.class)
public Result createVoucherOrder(Long voucherId) {
// 2、查询订单信息:
Long userId = UserHolder.getUser().getId();
long count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
if (count > 0) {
return Result.fail("您已购买过这张优惠券,不能重复购买");
}
// 3、购买优惠券,Stock更新,加入Order表
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1")
.eq("voucher_id", voucherId).gt("stock", 0).update();
if (!success) {
return Result.fail("库存不足");
}
VoucherOrder voucherOrder = new VoucherOrder();
Long orderId = redisIdWorker.nextId("order");
voucherOrder.setId(orderId);
voucherOrder.setUserId(UserHolder.getUser().getId());
voucherOrder.setVoucherId(voucherId);
save(voucherOrder);
return Result.ok(orderId);
}
2、如果直接在上述方法上加锁,锁的粒度太粗,不同的用户进入该方法时也会被锁住。因此在调用上述方法时,对userId.toString.intern()
加锁,保证相同的userId从常量池中拿到的数据为同一个对象。同时,为了使事务注解生效,需要调用代理对象 AopContext.currentProxy()
而不是该对象本身的方法。
@Override
public Result seckillVoucher(Long voucherId) {
// 1、查询优惠券信息
SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
LocalDateTime now = LocalDateTime.now();
LocalDateTime begin = seckillVoucher.getBeginTime();
LocalDateTime end = seckillVoucher.getEndTime();
if (now.isBefore(begin)) {
return Result.fail("秒杀尚未开始!");
}
if (now.isAfter(end)) {
return Result.fail("秒杀已经结束!");
}
int stock = seckillVoucher.getStock();
if (stock <= 0) {
return Result.fail("库存不足");
}
// 2、查询订单信息:
Long userId = UserHolder.getUser().getId();
synchronized (userId.toString().intern()) {
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
}
由于默认不可获得代理对象,需要在启动类上加入注释:
@EnableAspectJAutoProxy(exposeProxy = true)
并加入maven依赖:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>1.9.5</version>
</dependency>
3 分布式锁
满足分布式系统或集群模式下多进程可见并且互斥的锁。
使用MySQL比较少,Redis和Zookeeper比较常见。
3.1 Redis实现分布式锁
-
获取锁:
-
互斥:确保只能有一个线程获取锁
-
非阻塞:尝试一次,成功返回true,失败返回false
-
-
释放锁:
- 手动释放:只能释放属于该线程的锁
- 超时释放:获取锁时添加一个超时时间
@Component
public class SimpleRedisLock {
/**
* 标识这个锁
*/
private System name;
private static final String KEY_PREFIX = "lock:";
/**
* 由于不同的JVM可能有相同的线程号,所以需要ID_PREFIX来表示属于哪个服务,拼接threadId来唯一标识线程
*/
private static final String ID_PREFIX = UUID.randomUUID() + "-";
@Resource
StringRedisTemplate stringRedisTemplate;
public boolean tryLock(long timeoutSec) {
// 设置锁的值为获得当前锁的线程
long threadId = Thread.currentThread().threadId();
// 尝试获得锁,设置锁的过期时间以防止死锁
return Boolean.TRUE.equals(
stringRedisTemplate.opsForValue()
.setIfAbsent(KEY_PREFIX + name, ID_PREFIX + threadId,
timeoutSec, TimeUnit.SECONDS)
);
}
public void unlock() {
// 先判断当前线程有没有资格删掉这个锁,即redis中存储的线程id和当前线程id是否一致
String threadId = ID_PREFIX + Thread.currentThread().threadId();
String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
// 如果这个锁确实是当前服务器上 & 当前线程的锁
if (threadId.equals(id)) {
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
}
Long userId = UserHolder.getUser().getId();
SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
// 获取锁失败,已有线程进入该段逻辑
if (!lock.tryLock(1200)) {
return Result.fail("请勿重复下单");
}
try {
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
finally {
lock.unlock();
}
启动两个SpringBoot服务模拟分布式进行测试:
E:\leetcode\project_pre\Dianping\Front\conf\nginx.conf
# proxy_pass http://127.0.0.1:8081;
proxy_pass http://backend;
}
}
upstream backend {
server 127.0.0.1:8081 max_fails=5 fail_timeout=10s weight=1;
server 127.0.0.1:8082 max_fails=5 fail_timeout=10s weight=1;
}
- 可以使用LUA脚本进一步保证拿锁/还锁的原子性
3.2 Redisson实现分布式锁
3.2.1 maven依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
3.2.2 使用示例
@Test
void testRedisson() throws Exception{
// 创建锁对象
RLock lock = redissonClient.getLock("anyLock");
//尝试获取锁,参数分别是:获取锁的最大等待时间(期间会重试),锁自动释放时间,时间单位
boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
//判断获取锁成功
if(isLock){
try{
System.out.println("执行业务");
}finally{
//释放锁
lock.unlock();
}
}
}
ServiceImpl
RLock lock = redissonClient.getLock("order:" + userId);
// 获取锁失败,已有线程进入该段逻辑
if (!lock.tryLock()) {
return Result.fail("请勿重复下单");
}
try {
IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
return proxy.createVoucherOrder(voucherId);
}
finally {
lock.unlock();
}
3.3 Redisson可重入锁原理
逻辑如下右图:
method1调用method2,一个线程连续两次获取锁:重入。
在Lock锁中借助底层的一个voaltile
的state变量来记录重入的状态。
- 比如当前没有人持有这把锁,那么state=0
- 假如有人持有这把锁,那么state=1,如果持有这把锁的人再次持有这把锁,那么state就会+1 ,
- 释放一次就-1 ,直到减少成0 时,表示当前这把锁没有被人持有
采用hash结构用来存储锁,其中大key表示表示这把锁是否存在,用小key表示当前这把锁被哪个线程持有。
KEYS[1]
: 锁名称
ARGV[1]
: 锁失效时间
ARGV[2]
: id + ":" + threadId
,锁的小key
如果当前这把锁不存在,向redis中写一个hash数据并设置expire
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
如果这个锁已经存在,通过大key + 小key判断当前这把锁是否是属于自己的,如果是自己的,+1,重置锁时间
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);"
3.4 Redisson锁重试与WatchDog机制
waitTime,leaseTime
boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
第一个参数为重试等待时间,加入该参数以后,成为一个可重试的锁。
第二个参数为持有锁时间,默认为30s。
3.5 主从一致性问题
为了提高Redis的可用性,我们会搭建集群或者主从。
以主从为例:我们执行写命令,写在主机上, 主机会将数据同步给从机。但是假设在主机还没有来得及把数据写入到从机去的时候,此时主机宕机,哨兵会发现主机宕机,并且选举一个slave变成master,而此时新的master中实际上并没有锁信息,此时锁信息就丢失了。文章来源:https://www.toymoban.com/news/detail-539346.html
为了解决这个问题,Redisson提出来了MutiLock锁,每个节点的地位都是一样的, 加锁的逻辑需要写入到每一个主丛节点上,只有所有的服务器都写入成功,此时才是加锁成功。文章来源地址https://www.toymoban.com/news/detail-539346.html
到了这里,关于【Redis】秒杀业务设计、悲观锁与乐观锁的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!