1. 简介
随着技术的快速发展，业务系统规模的不断扩大，分布式系统越来越普及。一个应用往往会部署到多台机器上，在一些业务场景中，为了保证数据的一致性，要求在同一时刻，同一任务只在一个节点上运行，保证同一个方法同一时刻只能被一个线程执行。这时候分布式锁就运用而生了。
分布式锁有很多的解决方案。常见的有：
-
基于数据库的：悲观锁，乐观锁。
-
基于zookeeper的分布式锁。
-
本章中讲的基于redis的分布式锁。
2. 超卖
下单减库存是互联网项目中必不可少的环节。然而，如果我么考虑不得当，将会带来很多问题。比如最不能忍受的：超卖
如下代码，一个初始化库存的方法和一个购买图书的方法，我们没有做任何的并发处理，查看下最终结果。
package com.ldx.redisson.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.Objects;
/**
* redis 实现分布式锁
*
* @author ludangxin
* @date 2021/8/15
*/
@Slf4j
@RestController
@RequestMapping("redis")
public class RedisLockTestController {
@Resource
private StringRedisTemplate stringRedisTemplate;
// 商品key
private static final String KEY = "book";
// 库存数量
private static final Long STOCK = 50L;
/**
* 初始化
*/
@GetMapping("init")
public String init() {
stringRedisTemplate.opsForValue().set(KEY, String.valueOf(STOCK));
return "初始化成功～";
}
/**
* 购买图书
*/
@GetMapping("buy")
public String buy() {
// 获取到当前库存
String buyBefore = stringRedisTemplate.opsForValue().get(KEY);
if(Objects.isNull(buyBefore)) {
log.error("未找到\"{}\"的库存信息~", KEY);
return "暂未上架～";
}
long buyBeforeL = Long.parseLong(buyBefore);
if(buyBeforeL > 0) {
// 对库存进行-1操作
Long buyAfter = stringRedisTemplate.opsForValue().decrement(KEY);
log.info("剩余图书==={}", buyAfter);
return "购买成功～";
}
else {
log.info("库存不足～");
return "库存不足～";
}
}
}
启动测试：
? 这里我们使用jemter来进行并发请求。配置如下：
? 线程组配置：
? 请求配置：
? 请求结果：
只复制了部分日志
? 通过日志很明显的看到，即使在业务代码中判断了库存 > 0但还是超卖了。
......
2021-08-15 21:01:22.614 INFO 66913 --- [io-8080-exec-30] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.614 INFO 66913 --- [io-8080-exec-99] c.l.r.c.RedisLockTestController : 剩余图书===-42
2021-08-15 21:01:22.614 INFO 66913 --- [io-8080-exec-29] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.622 INFO 66913 --- [io-8080-exec-89] c.l.r.c.RedisLockTestController : 剩余图书===-40
2021-08-15 21:01:22.622 INFO 66913 --- [io-8080-exec-90] c.l.r.c.RedisLockTestController : 剩余图书===-35
2021-08-15 21:01:22.622 INFO 66913 --- [o-8080-exec-135] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.622 INFO 66913 --- [o-8080-exec-177] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.622 INFO 66913 --- [io-8080-exec-92] c.l.r.c.RedisLockTestController : 剩余图书===-34
2021-08-15 21:01:22.622 INFO 66913 --- [io-8080-exec-86] c.l.r.c.RedisLockTestController : 剩余图书===-37
2021-08-15 21:01:22.642 INFO 66913 --- [io-8080-exec-11] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.642 INFO 66913 --- [o-8080-exec-115] c.l.r.c.RedisLockTestController : 库存不足～
2021-08-15 21:01:22.642 INFO 66913 --- [io-8080-exec-72] c.l.r.c.RedisLockTestController : 剩余图书===-33
2021-08-15 21:01:22.643 INFO 66913 --- [nio-8080-exec-3] c.l.r.c.RedisLockTestController : 库存不足～
3. redis setnx
主要是用redis的 setnx （set not exists）命令实现分布式锁。
3.1 编写逻辑
在超买的场景中，我们了解了分布式锁的必要性。
上面的场景如果是单机的话，直接使用jvm锁就能解决问题，但是在分布式场景下下jvm锁无法处理。
接下来我们将使用redis命令来解决一下超卖问题。
-
新增了锁标识key。
-
在进行业务处理之前，给redis中
setIfAbsent(LOCK_KEY, clientId, 30, TimeUnit.SECONDS)作为lock。LOCK_KEY：锁的标识，比如秒杀的商品id_lock：当对该商品进行秒杀下单时，加锁使其线性执行。clientId：当前请求的唯一值，为了在删除锁时进行锁判断。即只能删除自己加的锁。防止误删锁。30：失效时间，防止死锁（如果加锁时不设置过期时间，当系统执行完加锁还未进行解锁时系统宕机，那其他节点也无法进行下单，因为锁一直在）。 -
解锁逻辑最好放在
finally中进行，防止报错导致死锁。
// 锁标识
private static final String LOCK_KEY = "book_lock";
/**
* 购买图书
*/
@GetMapping("buy1")
public String buy1() {
String clientId = UUID.randomUUID().toString();
/*
* 给redis设置一个key，并设置过期时间防止死锁
* setIfAbsent（setnx）：当key不存在时才设置值
* flag=true：值设置成功（获取锁） flag=false：设置值失败（获取锁失败）
*/
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(LOCK_KEY, clientId, 30, TimeUnit.SECONDS);
try {
if(Objects.nonNull(flag) && flag) {
String buyBefore = stringRedisTemplate.opsForValue().get(KEY);
if(Objects.isNull(buyBefore)) {
log.error("未找到\"{}\"的库存信息~", KEY);
return "暂未上架～";
}
long buyBeforeL = Long.parseLong(buyBefore);
if(buyBeforeL > 0) {
Long buyAfter = stringRedisTemplate.opsForValue().decrement(KEY);
log.info("剩余图书==={}", buyAfter);
return "购买成功～";
}
else {
log.info("库存不足～");
return "库存不足～";
}
}
else {
log.error("获取锁失败～");
}
}
catch(Exception e) {
e.printStackTrace();
}
finally {
// 防止误删锁
if(clientId.equals(stringRedisTemplate.opsForValue().get(LOCK_KEY))) {
stringRedisTemplate.delete(LOCK_KEY);
}
}
return "系统错误～";
}
3.2 启动测试
? 启动两个服务，并配置nginx负载均衡。
? nginx配置如下：
? jemter配置如下：
启动测试：
? 部分日志如下：
redis中查看库存：
打完收工～
3.3 小节
这里主要是用了setnx来实现分布式锁。虽然解决了超卖问题，但其中还是有很多缺陷。比如：
- 当请求获取锁失败时，能不能尝试重新获取锁或者阻塞等待获取锁，而不是直接返回
系统繁忙之类的提示语。 - 如果持有锁的请求处理时间超过了设置的超时时间，也就是业务逻辑还没有处理完呢，但锁已经失效了，此时刚好又进来一个请求，又有并发问题了??。
此时：redisson：申请出战??。
4. redisson
4.1 简介
Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。
Redisson底层采用的是Netty框架。支持Redis2.8以上版本，支持Java1.6+以上版本。
Jedis 与 Redisson
-
Jedis：Jedis 只是简单的封装了 Redis 的API库，可以看作是Redis客户端，它的方法和Redis 的命令很类似，相比于Redisson 更原生一些，更灵活。
-
Redisson：Redisson 不仅封装了 redis ，还封装了对更多数据结构的支持，以及锁等功能，相比于Jedis 更加大。
4.2 quick start
4.2.1 添加依赖
springboot 基础上添加此依赖。
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.16.1</version>
</dependency>
4.2.2 application.yaml
因为使用的是单机redis，并且使用的是自动装配的依赖，所以直接使用redis的基本配置即可。
server:
port: ${port}
spring:
# redis 配置
redis:
# 地址
host: localhost
# 端口，默认为6379
port: 6379
# 连接超时时间
timeout: 10s
lettuce:
pool:
# 连接池中的最小空闲连接
min-idle: 0
# 连接池中的最大空闲连接
max-idle: 8
# 连接池的最大数据库连接数
max-active: 8
# #连接池最大阻塞等待时间（使用负值表示没有限制）
max-wait: -1ms
4.2.3 controller
package com.ldx.redisson.controller;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.TimeUnit;
/**
* redisson test
*
* @author ludangxin
* @date 2021/8/15
*/
@Slf4j
@RestController
@RequestMapping("test")
@RequiredArgsConstructor
public class RedissonLockTestController {
private final RedissonClient redissonClient;
/**
* 没获取到锁的线程阻塞等待获取锁
*/
@GetMapping("/lock")
public void lock() {
log.info("进入了测试方法～");
RLock lock = null;
try {
lock = redissonClient.getLock("lock");
lock.lock();
log.info("获取到锁~");
Thread.sleep(2000);
}
catch(InterruptedException e) {
e.printStackTrace();
}
finally {
//如果当前线程保持锁定则解锁
if (null != lock && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
/**
* 没获取到锁的线程直接返回锁状态
*/
@GetMapping("tryLock")
public void tryLock() {
log.info("进入了测试方法～");
RLock lock = null;
try {
lock = redissonClient.getLock("lock");
if(lock.tryLock()) {
log.info("获取到锁~");
Thread.sleep(6000);
}
else {
log.error("获取锁失败~");
}
}
catch(InterruptedException e) {
e.printStackTrace();
}
finally {
//如果当前线程保持锁定则解锁
if (null != lock && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
/**
* 没获取到锁的线程尝试获取锁
*/
@GetMapping("tryLockWithBlock")
public void tryLockWithBlock() {
log.info("进入了测试方法～");
RLock lock = null;
try {
//非公平锁,随机取一个等待中的线程分配锁
lock = redissonClient.getLock("lock");
//公平锁,按照先后顺序依次分配锁
//lock=redissonClient.getFairLock("lock");
//最多等待锁3秒，5秒后强制解锁
if(lock.tryLock(3, 5, TimeUnit.SECONDS)) {
log.info("获取到锁~");
Thread.sleep(1000);
}
else {
log.error("获取锁失败~");
}
}
catch(InterruptedException e) {
e.printStackTrace();
}
finally {
//如果当前线程保持锁定则解锁
if (null != lock && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
}
4.2.4 启动测试
? jemter 配置如下：
? 启动9个线程并同一时刻进行请求。
? 请求lock方法日志如下：
? 所有请求同一时刻进入方法，并且请求阻塞每隔两秒获取到锁。
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-7] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.435 INFO 63602 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.437 INFO 63602 --- [io-8080-exec-10] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.437 INFO 63602 --- [io-8080-exec-11] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.437 INFO 63602 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:58:26.443 INFO 63602 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:28.474 INFO 63602 --- [io-8080-exec-11] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:30.499 INFO 63602 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:32.523 INFO 63602 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:34.548 INFO 63602 --- [io-8080-exec-10] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:36.569 INFO 63602 --- [nio-8080-exec-7] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:38.595 INFO 63602 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:40.616 INFO 63602 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:58:42.643 INFO 63602 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 获取到锁~
? 请求tryLock方法日志如下：
? 所有请求同一时刻进入方法，并且只有一个请求获取到了锁，其他请求直接返回结果。
2021-08-15 16:59:17.925 INFO 63602 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.925 INFO 63602 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.925 INFO 63602 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.928 INFO 63602 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.931 INFO 63602 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.931 INFO 63602 --- [io-8080-exec-11] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.933 INFO 63602 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.933 INFO 63602 --- [nio-8080-exec-8] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.933 INFO 63602 --- [io-8080-exec-10] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 16:59:17.937 ERROR 63602 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.937 INFO 63602 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 16:59:17.938 ERROR 63602 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.937 ERROR 63602 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.939 ERROR 63602 --- [nio-8080-exec-8] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.939 ERROR 63602 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.939 ERROR 63602 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.939 ERROR 63602 --- [io-8080-exec-11] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 16:59:17.939 ERROR 63602 --- [io-8080-exec-10] c.l.r.c.RedissonLockTestController : 获取锁失败~
? 请求tryLockWithBlock方法日志如下：
? 所有请求同一时刻进入方法，并且只有三个请求获取到了锁。
2021-08-15 23:34:41.617 INFO 70413 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.617 INFO 70413 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.617 INFO 70413 --- [nio-8080-exec-7] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.617 INFO 70413 --- [nio-8080-exec-8] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.618 INFO 70413 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.618 INFO 70413 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.618 INFO 70413 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.618 INFO 70413 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.617 INFO 70413 --- [nio-8080-exec-3] c.l.r.c.RedissonLockTestController : 进入了测试方法～
2021-08-15 23:34:41.658 INFO 70413 --- [nio-8080-exec-8] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 23:34:42.681 INFO 70413 --- [nio-8080-exec-7] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 23:34:43.697 INFO 70413 --- [nio-8080-exec-3] c.l.r.c.RedissonLockTestController : 获取到锁~
2021-08-15 23:34:44.624 ERROR 70413 --- [nio-8080-exec-9] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 23:34:44.624 ERROR 70413 --- [nio-8080-exec-1] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 23:34:44.624 ERROR 70413 --- [nio-8080-exec-5] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 23:34:44.624 ERROR 70413 --- [nio-8080-exec-2] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 23:34:44.625 ERROR 70413 --- [nio-8080-exec-6] c.l.r.c.RedissonLockTestController : 获取锁失败~
2021-08-15 23:34:44.630 ERROR 70413 --- [nio-8080-exec-4] c.l.r.c.RedissonLockTestController : 获取锁失败~
4.2.5 小节
redisson 提供了lock()和tryLock()，tryLock(long time, TimeUnit unit)，tryLock(long waitTime, long leaseTime, TimeUnit unit)方法。
-
lock()：会阻塞未获取锁的请求，默认持有30s锁，但当业务方法在30s内没有执行完时，会有看门狗（默认每隔10s）给当前锁续时30s。 -
tryLock()：尝试获取锁，获取不到则直接返回获取失败，默认持有30s锁，但当业务方法在30s内没有执行完时，会有看门狗（默认每隔10s）给当前锁续时30s。 -
tryLock(long time, TimeUnit unit)：尝试获取锁，等待time TimeUnit，默认持有30s锁，但当业务方法在30s内没有执行完时，会有看门狗（默认每隔10s）给当前锁续时30s。 -
tryLock(long waitTime, long leaseTime, TimeUnit unit)：尝试获取锁，等待waitTime TimeUnit，锁最长持有leaseTime TimeUnit，当业务方法在leaseTime TimeUnit时长内没有执行完时，会强制解锁。
4.3 解决超卖
private static final String KEY = "book";
/**
* 购买图书
*/
@GetMapping("buy1")
public String buy1() {
RLock lock = null;
try {
lock = redissonClient.getLock("lock");
if(lock.tryLock(3, TimeUnit.SECONDS)) {
RAtomicLong buyBefore = redissonClient.getAtomicLong(KEY);
if(Objects.isNull(buyBefore)) {
log.error("未找到\"{}\"的库存信息~", KEY);
return "暂未上架～";
}
long buyBeforeL = buyBefore.get();
if(buyBeforeL > 0) {
Long buyAfter = buyBefore.decrementAndGet();
log.info("剩余图书==={}", buyAfter);
return "购买成功～";
}
else {
log.info("库存不足～");
return "库存不足～";
}
}
else {
log.error("获取锁失败～");
}
}
catch(Exception e) {
e.printStackTrace();
}
finally {
//如果当前线程保持锁定则解锁
if(null != lock && lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
return "系统错误～";
}
经测试不会存在超卖问题。
并且避免了3.3小节中提到的问题。
4.4 小节
方便，好用。