问题背景
项目里刚好需要实现一个延迟订单取消任务。具体而言，如果一份订单在生成后的15分钟内未完成支付，系统需要自动取消该订单，并返还相关订单所使用的优惠券或免费额度等资源。
虽然引入MQ或者Kafka也是一种解决方法，但出于最大程度减少系统复杂性的角度考虑，强烈建议充分利用已有的Redis组件（例如Redisson）来解决这一问题，而不引入新组件。这样可以提高效率、减少维护负担，并确保充分发挥已有技术的潜力。
延迟队列
Redisson中定义了分布式延迟队列RDelayedQueue，是一种基于zset结构实现的延时队列，，它允许以指定的延迟时长，将任务放到目标队列中。
其实就是在zset的基础上增加了一个基于内存的延迟队列，当我们要添加一个数据到延迟队列的时候，Redission会把数据和超时时间放到zset中，并且起一个延时任务，当任务到期时，再去zset中把数据取出来，返回给客户端使用。
导入依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.2</version>
</dependency>编写配置类
@Configuration
public class RedisConfig {
@Value("${spring.redis.host}")
private String REDIS_HOST; //地址配置在配置文件上
@Value("${spring.redis.port}")
private String REDIS_PORT;
@Value("${spring.redis.password}")
private String REDIS_PASSWORD;
@Bean
public RedissonClient createRedisAPi(){
Config redissonConfig = new Config();
redissonConfig.setCodec(new org.redisson.client.codec.StringCodec());
//我这里单节点演示一下
SingleServerConfig singleServerConfig = redissonConfig.useSingleServer();
singleServerConfig.setAddress(String.format("redis://%s:%s", REDIS_HOST, REDIS_PORT));
singleServerConfig.setPassword(REDIS_PASSWORD);
//设置几号数据库
singleServerConfig.setDatabase(0);
singleServerConfig.setConnectTimeout(10000);
singleServerConfig.setConnectionPoolSize(300);
return Redisson.create(redissonConfig);
}
}延迟队列执行器
@Slf4j
public class DelayTaskQueueExecutor<T> {
final private RBlockingQueue<T> queue;
final private Thread msgLooper;
private final DelayTaskQueueExecutor.Processor<T> processor;
public interface Processor<T> {
void process(T task) throws InterruptedException;
}
public DelayTaskQueueExecutor(String threadName, RBlockingQueue<T> queue, DelayTaskQueueExecutor.Processor<T> processor) {
this.queue = queue;
this.processor = processor;
this.msgLooper = new Thread(this::looper);
this.msgLooper.setName(threadName);
this.msgLooper.start();
}
private void looper() {
while(true) {
try {
T task = queue.take();
processor.process(task);
} catch (InterruptedException e) {
break;
} catch (Exception e) {
log.error(String.format("TaskQueueExecutor %s run task exception",
this.msgLooper.getName()), e);
}
}
}
}服务类
@Slf4j
@Service
public class DelayQueueService {
@Resource
private OrderMapper orderMapper;
@Resource
private UserInfoMapper userInfoMapper;
@Resource
private BonusInfoMapper bonusInfoMapper;
@Autowired
private RedissonClient redisson;
private RDelayedQueue<String> delayedQueue;
@PostConstruct
public void initDelayQueue() {
RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("orderDelayQueue");
delayedQueue = redisson.getDelayedQueue(blockingQueue);
new DelayTaskQueueExecutor<>("ORDER_DELAY", blockingQueue, this::processOrder);
}
/**
* 将订单信息加入到延迟队列中，并设置TTL
*
* @param orderId 订单信息
*/
public void addToDelayQueue(String orderId) {
delayedQueue.offer(orderId, 2, TimeUnit.MINUTES);
}
@Transactional
public void processOrder(String orderId) {
Torder order = orderMapper.selectOne(Wrappers.lambdaQuery(Torder.class).eq(Torder::getOrderId, orderId));
if (order != null) {
Integer status = order.getStatus();
// 如果订单仍然是未支付状态，并且字数已经预扣了，则返还预扣的字数，并且关闭订单
if (status == Common.ORDER_ORIGINAL_STATUS) {
order.setStatus(Common.ORDER_CLOSED_STATUS);
String userId = order.getToken();
// 查询是否是登录用户
UserInfo userInfo = userInfoMapper.selectOneForUpdate(userId);
if (userInfo != null) {
UsedWordsBonusInfo wordsBonusInfo = usedWordsBonusInfoMapper.selectOne(Wrappers.lambdaQuery(UsedWordsBonusInfo.class).eq(UsedWordsBonusInfo::getOrderId, order.getOrderId()));
// 判断是否已经预扣了
if (wordsBonusInfo.getSelfWordsDeducted() == Common.ALREADY_DEDUCTION) {
userInfo.setRegisterBonusWords(userInfo.getRegisterBonusWords() + wordsBonusInfo.getUsedRegisterBonus());
userInfo.setInviteBonusWords(userInfo.getInviteBonusWords() + wordsBonusInfo.getUsedInviteBonus());
log.info("订单：{}，返还注册字数：{}，邀请字数：{}", orderId, wordsBonusInfo.getUsedRegisterBonus(), wordsBonusInfo.getUsedInviteBonus());
wordsBonusInfo.setSelfWordsDeducted(Common.ORIGIN_STATUS);
usedWordsBonusInfoMapper.updateById(wordsBonusInfo);
userInfoMapper.updateById(userInfo);
}
}
}
orderMapper.updateById(order);
}
}
}使用方法，创建订单的时候将订单号加入到队列中，到期后会自动执行关闭订单的对应逻辑
delayQueueService.addToDelayQueue(orderId);注意事项
细心的同学可能注意到了我这里的事务
java
@Transactional
public void processOrder(String orderId) {
Torder order = orderMapper.selectOne(Wrappers.lambdaQuery(Torder.class).eq(Torder::getOrderId, orderId));
if (order != null) {
Integer status = order.getStatus();
// 如果订单仍然是未支付状态，并且字数已经预扣了，则返还预扣的字数，并且关闭订单
if (status == Common.ORDER_ORIGINAL_STATUS) {
order.setStatus(Common.ORDER_CLOSED_STATUS);
String userId = order.getToken();
// 查询是否是登录用户
UserInfo userInfo = userInfoMapper.selectOneForUpdate(userId);
if (userInfo != null) {
UsedWordsBonusInfo wordsBonusInfo = usedWordsBonusInfoMapper.selectOne(Wrappers.lambdaQuery(UsedWordsBonusInfo.class).eq(UsedWordsBonusInfo::getOrderId, order.getOrderId()));
// 判断是否已经预扣了
if (wordsBonusInfo.getSelfWordsDeducted() == Common.ALREADY_DEDUCTION) {
userInfo.setRegisterBonusWords(userInfo.getRegisterBonusWords() + wordsBonusInfo.getUsedRegisterBonus());
userInfo.setInviteBonusWords(userInfo.getInviteBonusWords() + wordsBonusInfo.getUsedInviteBonus());
log.info("订单：{}，返还注册字数：{}，邀请字数：{}", orderId, wordsBonusInfo.getUsedRegisterBonus(), wordsBonusInfo.getUsedInviteBonus());
wordsBonusInfo.setSelfWordsDeducted(Common.ORIGIN_STATUS);
usedWordsBonusInfoMapper.updateById(wordsBonusInfo);
userInfoMapper.updateById(userInfo);
}
}
}
orderMapper.updateById(order);
}
}重点是我这里的
userInfoMapper.selectOneForUpdate(userId);，这句其实是我手写的一个SQL，SELECT XXX FROM user WHERE userId = xxx FOR UPDATE，主要是为了手动触发行锁。这里是使用的默认的事务隔离级别，可重复读。因为事务里的读操作默认是不会触发行锁的，所以这里可能会出现另一个事务将用户信息改了，并且提交了，由于可重复读的问题，当前事务中读取到的仍是修改前的数据，那么当前事务提交的时候，就会将另一个事务的提交结果覆盖掉，如果这里不触发行锁，会导致数据的不一致性。加了行锁之后，可以确保只有一个会话可以访问该订单数据，从而避免并发问题。但是也不是所有在事务里的读操作都要加行锁，毕竟那样的执行效率就太慢了，只有涉及到读取信息，并且后续需要对该信息进行修改的时候，才加上行锁。