Redis分布式锁(下篇）

╰+哭是因爲堅強的太久メ 2024-04-07 12:51 8阅读 0赞

我们在不久前介绍了SpringBoot定时任务，最近又一起探究了如何使用Redis实现简单的消息队列，都是一些不错的小知识点。为了能跟前面的内容产生联动，这次我们打算把Redis分布式锁相关的介绍融合进定时任务的案例中，学起来更带劲~

Redis的锁长啥样？

上一篇我们粗略介绍了JVM锁，比如synchronized关键字和ReentrantLock，它们都是实实在在已经实现的锁，而且还有标志位啥的。但Redis就是一个内存...怎么作为锁呢?

有一点大家要明确，Redis之所以能用来做分布式锁，肯定不只是因为它是一片内存，否则JVM本身也占有内存，为什么无法自己实现分布式锁呢？

我个人的理解是，要想自定义一个分布式锁，必须至少满足几个条件：

独立于多节点系统之外的一片内存  
唯一性（可以通过单线程，也可以通过选举机制，能保证唯一即可）  
当然，如果性能高一点，甚至支持高可用就更好啦  
以上三点Redis都能满足。在上面三个条件下，其实怎么设计锁，完全取决于个人如何定义锁。就好比现实生活中，通常我们理解的锁就是有个钥匙孔、需要插入钥匙的金属小物件。然而锁的形态可不止这么一种，随着科技的发展，什么指纹锁、虹膜锁层出不穷，但归根结底它们之所以被称为“锁”，是因为都保证了“唯一”。

如果我们能设计一种逻辑，它能造成某个场景下的“唯一事件”，那么它就可以被称为“锁”。比如，某家很有名的网红店，一天只接待一位客人。门口没有营业员，就放了一台取号机，里面放了一张票。你如果去迟了，票就没了，你就进不了这家店。这个场景下，没票的顾客进不去，被锁在门外。此时，取票机造成了“唯一事件”，那么它就可以叫做“锁”。

而Redis提供了setnx指令，如果某个key当前不存在则设置成功并返回true，否则不再重复设置，直接返回false。这不就是编程界的取号机吗？当然，实际用到的命令可不止这一个，具体如何实现，大家等下看代码即可。

Demo构思

在我看来，同样需要使用锁，动机可能完全相反：

在保证线程安全的前提下，尽量让所有线程都执行成功  
在保证线程安全的前提下，只让一个线程执行成功  
前者适用于秒杀等场景。作为商家，当然希望在不发生线程安全问题的前提下，让每一个订单都生效，直到商品售罄。此时分布式锁的写法可以是“不断重试”或“阻塞等待”，即：递归或while true循环尝试获取、阻塞等待。

![97626143a5634f31be3798b632d8ab89.png][]  
而后者适用于分布式系统或多节点项目的定时任务，比如同一份代码部署在A、B两台服务器上，而数据库共用同一个。如果不做限制，那么在同一时刻，两台服务器都会去拉取列表执行，会发生任务重复执行的情况。

![e9b0043d29234aca8c340a8ff01ef8f5.png][]  
此时可以考虑使用分布式锁，在cron触发的时刻只允许一个线程去往数据库拉取任务：

![9164d6e331dd40339fb1d59c5f300a97.png][]  
在实现Redis分布式锁控制定时任务唯一性的同时，我们引入之前的Redis消息队列。注意，这与Redis分布式锁本身无关，就是顺便复习一遍Redis消息队列而已，大家可以只实现Redis分布式锁+定时任务的部分。

整个Demo的结构大致如图：

![3895a726d7e4457fbf035ee34823d0cb.png][]  
当然，实际项目中一般是这样的：

![8cd2164498404d778069057e7cb3c673.png][]  
分布式锁为什么难设计？

首先，要和大家说一下，但凡牵涉到分布式的处理，没有一个是简单的，上面的Demo设计也不过是玩具，用来启发 大家的思路。

为什么要把demo设计得这么复杂呢？哈哈，因为这是我在上一家公司自己设计的，遇到了很多坑...拿出来自嘲一番，与各位共勉。

我当时的设计思路是：

由于小公司没有用什么Elastic-Job啥的，就是很普通的多节点部署。为了避免任务重复执行，我想设计一个分布式锁。但因为当时根本不知道redisson，所以就自己百度了redis实现分布式锁的方式，然后依葫芦画瓢自己手写了一个 。

但我写完redis分布式锁后，在实际测试过程中发现还需要考虑锁的失效时间...

这里有两个问题：

为什么要设置锁的过期时间？  
锁的过期时间设置多久合适？  
最简单的实现方案是这样的，一般没问题：

![e9a7668525564a0b9938e8b23540e273.png][]  
但极端的情况下（项目在任务进行时重启或意外宕机），可能当前任务来不及解锁就挂了（死锁），那么下一个任务就会一直被锁在方法外等待。就好比厕所里有人被熏晕了，没法开门，而外面的人又进不去...

![27663cad568b48369b6422f6bd496b8e.png][]  
此时需要装一个自动解锁的门，到时间自动开门，也就是要给锁设置一个过期时间。但紧接着又会有第二个问题：锁的失效时间设多长合适？

很难定。

因为随着项目的发展，定时任务的执行时间很可能是变化的。

如果设置时间过长，极端点，定为365天。假设任务正常执行，比如10分钟就结束，此时执行完毕的任务自己会主动解锁。但万一和上面一样宕机了，虽说你设置了过期时间，但下一个任务需要等一年才能执行...本质上和没有设置过期时间一样！就好比...你自己想想什么例子合适，能加深你的理解哦。

如果设置时间过短，上一个人还没拉完，门就“咔嚓”一声开了，尴尬不，重复执行了。

终上所述，我当时之所以设计得这么复杂，就是想尽量缩短任务执行的时间，让它尽可能短（拉取后直接丢给队列，自己不处理），这样锁的时间一般设置30分钟就没啥问题。另外，对于死锁问题，我当时没有考虑宕机的情况，只考虑了意外重启…问题还有很多，文末会再总结。

请大家阅读下面代码时思考两个问题：

Demo如何处理锁的过期时间  
Demo如何防止死锁  
项目搭建

新建一个空的SpringBoot项目。

拷贝下方代码，构建工程：

![bc159f0c89f34e43b1188076c31c56fc.png][]  
构建完以后，拷贝一份，修改端口号为8081，避免和原先的冲突

![021d03fec0d44e919464432de0a51e62.png][]  
统一管理Redis Key：RedisKeyConst

/\*\*  
\* 统一管理Redis Key  
\*  
\* @author qiyu  
\*/  
public abstract class RedisKeyConst \{  
/\*\*  
\* 分布式锁的KEY  
\*/  
public static final String RESUME\_PULL\_TASK\_LOCK = "resume\_pull\_task\_lock";  
/\*\*  
\* 简历异步解析任务队列  
\*/  
public static final String RESUME\_PARSE\_TASK\_QUEUE = "resume\_parse\_task\_queue";  
\}  
Redis消息队列：RedisMessageQueueConsumer

/\*\*  
\* 消费者，异步获取简历解析结果并存入数据库  
\*  
\* @author qiyu  
\*/  
@Slf4j  
@Component  
public class RedisMessageQueueConsumer implements ApplicationListener<ContextRefreshedEvent> \{

@Autowired  
private RedisService redisService;  
@Autowired  
private AsyncResumeParser asyncResumeParser;  
@Autowired  
private ObjectMapper objectMapper;

@Override  
public void onApplicationEvent(ContextRefreshedEvent event) \{  
log.info("开始监听RedisMessageQueue...");  
CompletableFuture.runAsync(() -> \{  
// 大循环，不断监听队列任务（阻塞式）  
while (true) \{  
// 阻塞监听  
ResumeCollectionDTO resumeCollectionDTO = (ResumeCollectionDTO) redisService.popQueue(RedisKeyConst.RESUME\_PARSE\_TASK\_QUEUE, 5, TimeUnit.SECONDS);  
if (resumeCollectionDTO != null) \{  
int rePullCount = 0;  
int retryCount = 0;  
log.info("从队列中取出:\{\}", resumeCollectionDTO.getName());  
log.info(">>>>>>>>>>>>>>>>>>>开始拉取简历:\{\}", resumeCollectionDTO.getName());  
Long asyncPredictId = resumeCollectionDTO.getAsyncPredictId();  
// 小循环，针对每一个任务多次调用第三方接口，直到获取最终结果或丢弃任务  
while (true) \{  
try \{  
PredictResult result = asyncResumeParser.getResult(asyncPredictId);  
rePullCount++;  
// 如果已经解析完毕  
if (result.getStatus() == 2) \{  
// 保存数据库  
try \{  
log.info("简历:\{\}解析成功", resumeCollectionDTO.getName());  
log.info("resultJson:\{\}", result.getResultJson());  
ResumeCollectionDO resumeCollectionDO = objectMapper.readValue(result.getResultJson(), ResumeCollectionDO.class);  
log.info("<<<<<<<<<<<<<<<<<<<保存简历:\{\}到数据库", resumeCollectionDO);  
// 归零  
rePullCount = 0;  
retryCount = 0;  
break;  
\} catch (Exception e) \{  
discardTask(resumeCollectionDTO);  
log.info("<<<<<<<<<<<<<<<<<<<保存简历失败，丢弃任务");  
rePullCount = 0;  
retryCount = 0;  
break;  
\}  
\}  
// 远程服务还未解析完毕，重试  
else \{  
try \{  
if (rePullCount <= 3) \{  
// 前3次重试，时间为1s间隔  
TimeUnit.SECONDS.sleep(1);  
log.info("简历:\{\}尚未解析完毕, 准备进行第\{\}次重试, 停顿1s后进行", resumeCollectionDTO.getName(), rePullCount);  
\} else if (rePullCount > 3 && rePullCount <= 6) \{  
// 说明任务比较耗时，加长等待时间  
TimeUnit.SECONDS.sleep(2);  
log.info("简历:\{\}尚未解析完毕, 准备进行第\{\}次重试, 停顿2s后进行", resumeCollectionDTO.getName(), rePullCount);  
\} else if (rePullCount > 6 && rePullCount <= 8) \{  
// 说明任务比较耗时，加长等待时间  
TimeUnit.SECONDS.sleep(3);  
log.info("简历:\{\}尚未解析完毕, 准备进行第\{\}次重试, 停顿3s后进行", resumeCollectionDTO.getName(), rePullCount);  
\} else \{  
discardTask(resumeCollectionDTO);  
log.info("<<<<<<<<<<<<<<<<<<<多次拉取仍未得到结果, 丢弃简历:\{\}", resumeCollectionDTO.getName());  
retryCount = 0;  
rePullCount = 0;  
break;  
\}  
\} catch (InterruptedException e) \{  
discardTask(resumeCollectionDTO);  
log.info("<<<<<<<<<<<<<<<<<<<任务中断异常, 简历:\{\}", resumeCollectionDTO.getName());  
rePullCount = 0;  
retryCount = 0;  
break;  
\}  
\}  
\} catch (Exception e) \{  
if (retryCount > 3) \{  
discardTask(resumeCollectionDTO);  
log.info("<<<<<<<<<<<<<<<<<<<简历:\{\}重试\{\}次后放弃, rePullCount:\{\}, retryCount:\{\}", resumeCollectionDTO.getName(), retryCount, rePullCount, retryCount);  
rePullCount = 0;  
retryCount = 0;  
break;  
\}  
retryCount++;  
log.info("简历:\{\}远程调用异常, 准备进行第\{\}次重试...", resumeCollectionDTO.getName(), retryCount);  
\}  
\}  
log.info("break......");  
\}  
\}  
\});  
\}

private void discardTask(ResumeCollectionDTO task) \{  
// 根据asyncPredictId删除任务...  
log.info("丢弃任务:\{\}...", task.getName());  
\}

\}  
实体类：DO+DTO

@Data  
@NoArgsConstructor  
@AllArgsConstructor  
public class ResumeCollectionDO \{  
/\*\*  
\* 简历id  
\*/  
private Long id;  
/\*\*  
\* 简历名称  
\*/  
private String name;  
\}

@Data  
@NoArgsConstructor  
@AllArgsConstructor  
public class ResumeCollectionDTO implements Serializable \{  
/\*\*  
\* 简历id  
\*/  
private Long id;  
/\*\*  
\* 异步解析id，稍后根据id可获取最终解析结果  
\*/  
private Long asyncPredictId;  
/\*\*  
\* 简历名称  
\*/  
private String name;  
\}  
分布式锁：RedisService

public interface RedisService \{

/\*\*  
\* 向队列插入消息  
\*  
\* @param queue 自定义队列名称  
\* @param obj 要存入的消息  
\*/  
void pushQueue(String queue, Object obj);

/\*\*  
\* 从队列取出消息  
\*  
\* @param queue 自定义队列名称  
\* @param timeout 最长阻塞等待时间  
\* @param timeUnit 时间单位  
\* @return  
\*/  
Object popQueue(String queue, long timeout, TimeUnit timeUnit);

/\*\*  
\* 尝试上锁  
\*  
\* @param lockKey  
\* @param value  
\* @param expireTime  
\* @param timeUnit  
\* @return  
\*/  
boolean tryLock(String lockKey, String value, long expireTime, TimeUnit timeUnit);

/\*\*  
\* 根据MACHINE\_ID解锁（只能解自己的）  
\*  
\* @param lockKey  
\* @param value  
\* @return  
\*/  
boolean unLock(String lockKey, String value);

/\*\*  
\* 释放锁，不管是不是自己的  
\*  
\* @param lockKey  
\* @param value  
\* @return  
\*/  
boolean releaseLock(String lockKey, String value);

@Slf4j  
@Component  
public class RedisServiceImpl implements RedisService \{

@Autowired  
private RedisTemplate redisTemplate;

/\*\*  
\* 向队列插入消息  
\*  
\* @param queue 自定义队列名称  
\* @param obj 要存入的消息  
\*/  
@Override  
public void pushQueue(String queue, Object obj) \{  
redisTemplate.opsForList().leftPush(queue, obj);  
\}

/\*\*  
\* 从队列取出消息  
\*  
\* @param queue 自定义队列名称  
\* @param timeout 最长阻塞等待时间  
\* @param timeUnit 时间单位  
\* @return  
\*/  
@Override  
public Object popQueue(String queue, long timeout, TimeUnit timeUnit) \{  
return redisTemplate.opsForList().rightPop(queue, timeout, timeUnit);  
\}

/\*\*  
\* 尝试上锁  
\*  
\* @param lockKey  
\* @param value  
\* @param expireTime  
\* @param timeUnit  
\* @return  
\*/  
@Override  
public boolean tryLock(String lockKey, String value, long expireTime, TimeUnit timeUnit) \{  
Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockKey, value);  
if (lock != null && lock) \{  
redisTemplate.expire(lockKey, expireTime, timeUnit);  
return true;  
\} else \{  
return false;  
\}  
\}

/\*\*  
\* 根据MACHINE\_ID解锁（只能解自己的）  
\*  
\* @param lockKey  
\* @param value  
\* @return  
\*/  
@Override  
public boolean unLock(String lockKey, String value) \{  
String machineId = (String) redisTemplate.opsForValue().get(lockKey);  
if (StringUtils.isNotEmpty(machineId) && machineId.equals(value)) \{  
redisTemplate.delete(lockKey);  
return true;  
\}  
return false;  
\}

/\*\*  
\* 释放锁，不管是不是自己的  
\*  
\* @param lockKey  
\* @param value  
\* @return  
\*/  
@Override  
public boolean releaseLock(String lockKey, String value) \{  
Boolean delete = redisTemplate.delete(lockKey);  
if (delete != null && delete) \{  
log.info("Spring启动，节点:\{\}成功释放上次简历汇聚定时任务锁", value);  
return true;  
\}  
return false;  
\}

\}  
定时任务：ResumeCollectionTask

@Slf4j  
@Component  
@EnableScheduling  
public class ResumeCollectionTask implements ApplicationListener<ContextRefreshedEvent> \{

/\*\*  
\* 当这份代码被部署到不同的服务器，启动时为每台机器分配一个唯一的机器ID  
\*/  
private static String MACHINE\_ID = IdUtil.randomUUID();

@Autowired  
private RedisService redisService;  
@Autowired  
private AsyncResumeParser asyncResumeParser;

@Scheduled(cron = "0 \*/1 \* \* \* ?")  
// @Scheduled(fixedDelay = 60 \* 1000L)  
public void resumeSchedule() \{  
// 尝试上锁，返回true或false，锁的过期时间设置为10分钟（实际要根据项目调整，这也是自己实现Redis分布式锁的难点之一）  
boolean lock = redisService.tryLock(RedisKeyConst.RESUME\_PULL\_TASK\_LOCK, MACHINE\_ID, 10, TimeUnit.MINUTES);

// 如果当前节点成功获取锁，那么整个系统只允许当前程序去MySQL拉取待执行任务  
if (lock) \{  
log.info("节点:\{\}获取锁成功，定时任务启动", MACHINE\_ID);  
try \{  
collectResume();  
\} catch (Exception e) \{  
log.info("定时任务异常:", e);  
\} finally \{  
redisService.unLock(RedisKeyConst.RESUME\_PULL\_TASK\_LOCK, MACHINE\_ID);  
log.info("节点:\{\}释放锁，定时任务结束", MACHINE\_ID);  
\}  
\} else \{  
log.info("节点:\{\}获取锁失败，放弃定时任务", MACHINE\_ID);  
\}  
\}

/\*\*  
\* 任务主体：  
\* 1.从数据库拉取符合条件的HR邮箱  
\* 2.从HR邮箱拉取附件简历  
\* 3.调用远程服务异步解析简历  
\* 4.插入待处理任务到数据库，作为记录留存  
\* 5.把待处理任务的id丢到Redis Message Queue，让Consumer去异步处理  
\*/  
private void collectResume() throws InterruptedException \{  
// 跳过1、2两步，假设已经拉取到简历  
log.info("节点:\{\}从数据库拉取任务简历", MACHINE\_ID);  
List<ResumeCollectionDO> resumeCollectionList = new ArrayList<>();  
resumeCollectionList.add(new ResumeCollectionDO(1L, "张三的简历.pdf"));  
resumeCollectionList.add(new ResumeCollectionDO(2L, "李四的简历.html"));  
resumeCollectionList.add(new ResumeCollectionDO(3L, "王五的简历.doc"));  
// 模拟数据库查询耗时  
TimeUnit.SECONDS.sleep(3);

log.info("提交任务到消息队列:\{\}", resumeCollectionList.stream().map(ResumeCollectionDO::getName).collect(Collectors.joining(",")));

for (ResumeCollectionDO resumeCollectionDO : resumeCollectionList) \{  
// 上传简历异步解析，得到异步结果id  
Long asyncPredictId = asyncResumeParser.asyncParse(resumeCollectionDO);

// 把任务插入数据库  
// 略...

// 把任务丢到Redis Message Queue  
ResumeCollectionDTO resumeCollectionDTO = new ResumeCollectionDTO();  
BeanUtils.copyProperties(resumeCollectionDO, resumeCollectionDTO);  
resumeCollectionDTO.setAsyncPredictId(asyncPredictId);  
redisService.pushQueue(RedisKeyConst.RESUME\_PARSE\_TASK\_QUEUE, resumeCollectionDTO);  
\}

/\*\*  
\* 项目重启后先尝试删除之前的锁（如果存在），防止死锁等待  
\*  
\* @param event the event to respond to  
\*/  
@Override  
public void onApplicationEvent(ContextRefreshedEvent event) \{  
redisService.releaseLock(RedisKeyConst.RESUME\_PULL\_TASK\_LOCK, MACHINE\_ID);  
\}

\}  
模拟第三方服务（异步）

/\*\*  
\* 第三方提供给的简历解析服务  
\*  
\* @author qiyu  
\*/  
@Service  
public class AsyncResumeParser \{

@Autowired  
private ObjectMapper objectMapper;

/\*\*  
\* 模拟分配异步任务结果id，不用深究，没啥意义，反正每个任务都会得到一个id，稍后根据id返回最终解析结果  
\*/  
private static final AtomicLong ASYNC\_RESULT\_ID = new AtomicLong(1000);  
/\*\*  
\* 解析结果  
\*/  
private static final Map<Long, String> results = new HashMap<>();

/\*\*  
\* 模拟第三方服务异步解析，返回解析结果  
\*  
\* @param resumeCollectionDO  
\* @return  
\*/  
public Long asyncParse(ResumeCollectionDO resumeCollectionDO) \{  
long asyncPredictId = ASYNC\_RESULT\_ID.getAndIncrement();  
try \{  
String resultJson = objectMapper.writeValueAsString(resumeCollectionDO);  
results.put(asyncPredictId, resultJson);  
return asyncPredictId;  
\} catch (JsonProcessingException e) \{  
e.printStackTrace();  
\}  
return -1L;  
\}

/\*\*  
\* 根据异步id返回解析结果，但此时未必已经解析成功  
\* <p>  
\* 解析状态  
\* 0 初始化  
\* 1 处理中  
\* 2 调用成功  
\* 3 调用失败  
\*  
\* @param asyncPredictId  
\* @return  
\*/  
public PredictResult getResult(Long asyncPredictId) throws ParseErrorException, InterruptedException \{  
// 随机模拟异步解析的状态  
int value = ThreadLocalRandom.current().nextInt(100);  
if (value >= 85) \{  
// 模拟解析完成  
TimeUnit.SECONDS.sleep(1);  
String resultJson = results.get(asyncPredictId);  
return new PredictResult(resultJson, 2);  
\} else if (value <= 5) \{  
// 模拟解析异常  
TimeUnit.SECONDS.sleep(1);  
throw new ParseErrorException("简历解析异常");  
\}  
// 如果时间过短，返回status=1，表示解析中  
TimeUnit.SECONDS.sleep(1);  
return new PredictResult("", 1);  
\}

/\*\*  
\* 解析异常  
\*  
\* @author qiyu  
\*/  
public class ParseErrorException extends Exception \{  
/\*\*  
\* Constructs a new exception with \{@code null\} as its detail message.  
\* The cause is not initialized, and may subsequently be initialized by a  
\* call to \{@link \#initCause\}.  
\*/  
public ParseErrorException() \{  
\}

/\*\*  
\* Constructs a new exception with the specified detail message. The  
\* cause is not initialized, and may subsequently be initialized by  
\* a call to \{@link \#initCause\}.  
\*  
\* @param message the detail message. The detail message is saved for  
\* later retrieval by the \{@link \#getMessage()\} method.  
\*/  
public ParseErrorException(String message) \{  
super(message);  
\}  
\}

/\*\*  
\* 第三方返回值  
\*  
\* @author qiyu  
\*/  
@Data  
@NoArgsConstructor  
@AllArgsConstructor  
public class PredictResult \{  
/\*\*  
\* 解析结果  
\*/  
private String resultJson;  
/\*\*  
\* 解析状态  
\* 0 初始化  
\* 1 处理中  
\* 2 调用成功  
\* 3 调用失败  
\*/  
private Integer status;  
\}  
模拟异常

在项目运行过程中，启动这个测试类的方法，即可观察不一样的现象。

@SpringBootTest  
class RedisDistributedLockApplicationTests \{

@Autowired  
private RedisService redisService;

/\*\*  
\* 作为失败案例（因为不存在777L这个解析任务，AsyncResumeParse.results会返回null）  
\* 观察RedisMessageQueueConsumer的处理方式  
\*/  
@Test  
void contextLoads() \{  
ResumeCollectionDTO resumeCollectionDTO = new ResumeCollectionDTO();  
resumeCollectionDTO.setId(666L);  
resumeCollectionDTO.setAsyncPredictId(777L);  
resumeCollectionDTO.setName("测试1号");

redisService.pushQueue(RedisKeyConst.RESUME\_PARSE\_TASK\_QUEUE, resumeCollectionDTO);

\}  
pom.xml

server:  
port: 8080

效果展示

啥都不说了，都在代码里了。大家自己拷贝到本地，动手玩一下，加深对Redis锁和Redis消息队列的理解。

![dcbd660dfb2c4559872c9a3738ee9a2f.png][]  
只有一个定时任务能去数据库拉取任务，到时多节点部署大致是下面这样（redis一般是独立部署的，和节点代码无关）：

![fa521fa1474a4110a4b57a0cbdff3ea0.png][]  
后话

上面展示的代码其实存在很多问题，我们会在下一篇指出并讨论解决方案。

本文仅提供思路，开阔大家的眼界，千万别在自己项目中使用！！！！我当年被这个坑惨了，花里胡哨的，尤其Consumer里一大堆的sleep()，是非常low的！！

对于异步调用的结果，不要循环等待，而应该分为几步：

调用异步接口，得到异步结果唯一id  
将结果id保存到任务表中，作为一个任务  
启动定时任务，根据id拉取最终结果（如果还没有结果，不更改状态，等下一个定时任务处理）  
分布式定时任务可以考虑xxl-job或elastic-job，分布式锁推荐使用redisson。![3db721cfc37443588006580392f776a2.gif][]

[97626143a5634f31be3798b632d8ab89.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/c2a6d3b22c7b4a49a3a4b48729389300.png
[e9b0043d29234aca8c340a8ff01ef8f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/e4747cb31e004a6d871049c211fc1c07.png
[9164d6e331dd40339fb1d59c5f300a97.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/0a041c6c76ba44ca9673cf54f68b4e93.png
[3895a726d7e4457fbf035ee34823d0cb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/a076cb09ba564c65b0ef0d664e347986.png
[8cd2164498404d778069057e7cb3c673.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/483dd25bbf4b423bb3dcb98f7614081e.png
[e9a7668525564a0b9938e8b23540e273.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/3d4fe55d43784c11a52c67381da71447.png
[27663cad568b48369b6422f6bd496b8e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/4eca8f0a2d51466db7ca034d3582633d.png
[bc159f0c89f34e43b1188076c31c56fc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/f35eb343265b428b900a8d62b5faf5cc.png
[021d03fec0d44e919464432de0a51e62.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/4fb3be0f7c0f4704b28cd35516c7172a.png
[dcbd660dfb2c4559872c9a3738ee9a2f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/f4847b6efccf425493a80a05e2545866.png
[fa521fa1474a4110a4b57a0cbdff3ea0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/3163f5d32aae4acea5673679c0d5bb28.png
[3db721cfc37443588006580392f776a2.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/07/f98596182645421592aef602bf594516.gif