作者简介：大家好，我是smart哥，前中兴通讯、美团架构师，现某互联网公司CTO
联系qq：184480602，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬
我们在不久前介绍了SpringBoot定时任务，最近又一起探究了如何使用Redis实现简单的消息队列，都是一些不错的小知识点。为了能跟前面的内容产生联动，这次我们打算把Redis分布式锁相关的介绍融合进定时任务的案例中，学起来更带劲~
Demo构思
在我看来，同样需要使用锁，动机可能完全相反：
- 在保证线程安全的前提下，尽量让所有线程都执行成功
- 在保证线程安全的前提下，只让一个线程执行成功
前者适用于秒杀等场景。作为商家，当然希望在不发生线程安全问题的前提下，让每一个订单都生效，直到商品售罄。此时分布式锁的写法可以是“不断重试”或“阻塞等待”，即：递归或while true循环尝试获取、阻塞等待。
而后者适用于分布式系统或多节点项目的定时任务，比如同一份代码部署在A、B两台服务器上，而数据库共用同一个。如果不做限制，那么在同一时刻，两台服务器都会去拉取列表执行，会发生任务重复执行的情况。
此时可以考虑使用分布式锁，在cron触发的时刻只允许一个线程去往数据库拉取任务：
在实现Redis分布式锁控制定时任务唯一性的同时，我们引入之前的Redis消息队列。注意，这与Redis分布式锁本身无关，就是顺便复习一遍Redis消息队列而已，大家可以只实现Redis分布式锁+定时任务的部分。
整个Demo的结构大致如图：
当然，实际项目中一般是这样的：
分布式锁为什么难设计？
首先，要和大家说一下，但凡牵涉到分布式的处理，没有一个是简单的，上面的Demo设计也不过是玩具，用来启发 大家的思路。
为什么要把Demo设计得这么复杂呢？哈哈，因为这是我在上一家公司自己设计的，遇到了很多坑...拿出来自嘲一番，与各位共勉。
我当时的设计思路是：
由于小公司没有用什么Elastic-Job啥的，就是很普通的多节点部署。为了避免任务重复执行，我想设计一个分布式锁。但因为当时根本不知道Redisson，所以就自己百度了Redis实现分布式锁的方式，然后依葫芦画瓢自己手写了一个 。
但我写完Redis分布式锁后，在实际测试过程中发现还需要考虑锁的失效时间...
这里有两个问题：
- 为什么要设置锁的过期时间？
- 锁的过期时间设置多久合适？
最简单的实现方案是这样的，一般没问题：
但极端的情况下（项目在任务进行时重启或意外宕机），可能当前任务来不及解锁就挂了（死锁），那么下一个任务就会一直被锁在方法外等待。就好比厕所里有人被熏晕了，没法开门，而外面的人又进不去...
此时需要装一个自动解锁的门，到时间自动开门，也就是要给锁设置一个过期时间。但紧接着又会有第二个问题：锁的失效时间设多长合适？
很难定。
因为随着项目的发展，定时任务的执行时间很可能是变化的。
如果设置时间过长，极端点，定为365天。假设任务正常执行，比如10分钟就结束，那么线程继续往下就会执行unLock()主动解锁。但万一和上面一样宕机了，那么这个锁就要等365天后才解开。注意，宕机可不像JVM异常，它压根不会去执行finally里的unLock()。这种情况好比有个人在厕所里上大号直接掉坑里了，而自动门默认365天打开…所以，锁过期时间设置过长的坏处，本质是一旦发生宕机来不及解锁，那么过期时间越长，影响面越广，会导致其他操作阻滞。
如果设置时间过短，上一个人还没拉完，门就“咔嚓”一声开了，尴尬不，重复执行了。
终上所述，我当时之所以设计得这么复杂，就是想尽量缩短任务执行的时间，让它尽可能短（拉取后直接丢给队列，自己不处理），这样锁的时间一般设置30分钟就没啥问题。另外，对于死锁问题，我当时没有考虑宕机的情况，只考虑了意外重启…问题还有很多，文末会再总结。
请大家阅读下面代码时思考两个问题：
- Demo如何处理锁的过期时间
- Demo如何防止死锁
项目搭建
新建一个空的SpringBoot项目。
拷贝下方代码，构建工程：
构建完以后，拷贝一份，修改端口号为8081，避免和原先的冲突
统一管理Redis Key：RedisKeyConst
/**
* 统一管理Redis Key
*
* @author mx
*/
public final 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 mx
*/
@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 (Boolean.TRUE.equals(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 (Boolean.TRUE.equals(delete)) {
log.info("Spring启动，节点:{}成功释放上次简历汇聚定时任务锁", value);
return true;
}
return false;
}
}定时任务：ResumeCollectionTask
@Slf4j
@Component
@EnableScheduling
public class ResumeCollectionTask implements ApplicationListener<ContextRefreshedEvent> {
/**
* 当这份代码被部署到不同的服务器，启动时为每台机器分配一个唯一的机器ID
*/
private static final 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 mx
*/
@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 mx
*/
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 mx
*/
@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
spring:
redis:
host:
password:
database: 2效果展示
啥都不说了，都在jiu代码里了。大家自己拷贝到本地，动手玩一下，加深对Redis锁和Redis消息队列的理解。
只有一个定时任务能去数据库拉取任务，到时多节点部署大致是下面这样（redis一般是独立部署的，和节点代码无关）：
后话
上面展示的代码其实存在很多问题，我们会在下一篇指出并讨论解决方案。
本文仅提供思路，开阔大家的眼界，千万别在自己项目中使用！！！！我当年被这个坑惨了，花里胡哨的，尤其Consumer里一大堆的sleep()，是非常low的！！
对于异步调用的结果，不要循环等待，而应该分为几步：
- 调用异步接口，得到异步结果唯一id
- 将结果id保存到任务表中，作为一个任务
- 启动定时任务，根据id拉取最终结果（如果还没有结果，跳过当前任务，等下一个定时任务处理）
分布式定时任务可以考虑xxl-job或elastic-job，分布式锁推荐使用redisson。
作者简介：大家好，我是smart哥，前中兴通讯、美团架构师，现某互联网公司CTO
进群，大家一起学习，一起进步，一起对抗互联网寒冬
