redis实现延时队列_redis实现延迟队列 java

import com.greenutility.run.util.DateUtils;
import com.greenutility.run.util.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
 
import javax.annotation.Resource;
import java.util.Date;
import java.util.Set;
 
/**
 * redis zset 实现延时队列demo
 */
 
@Slf4j
@Component
public class OrderDelayService implements InitializingBean {
 
    public static final String DELAY_TASK_KEY = "delayTask:";
 
    @Resource
    private RedisUtils redisUtils;
 
    /**
     * 加入延时队列
     * @param id 任务id
     * @param time 延时时间（单位：分钟）
     */
    public void produce(String id, int time) {
        redisUtils.zsetAdd(
                DELAY_TASK_KEY,
                id,
                System.currentTimeMillis() + (time * 60 * 1000)
        );
        log.info("任务 " + id + " 加入延时队列成功，当前时间 = " + DateUtils.dateToString(new Date(), DateUtils.FULL));
    }
 
    /**
     * 从延时队列移除
     * @param id 任务id
     */
    public Boolean remove(String id) {
        log.info("任务 " + id + " 移除延时队列");
        return redisUtils.zsetRemove(DELAY_TASK_KEY, id);
    }
 
    //延时任务，也是异步任务，延时任务达到时效之后关闭订单，并将延时任务从redis zset删除
    @Async("test")
    public void consuming() {
        Set<ZSetOperations.TypedTuple<Object>> ids = redisUtils.zsetRangeList(
                DELAY_TASK_KEY,
                0,  //延时任务score最小值
                System.currentTimeMillis() //延时任务score最大值（当前时间）
        );
        if (!CollectionUtils.isEmpty(ids)) {
            for (ZSetOperations.TypedTuple<Object> id : ids) {
                log.info("任务 " + id.getValue() + " 超时被自动关闭, 关闭时间 = " + DateUtils.dateToString(new Date(), DateUtils.FULL));
                //todo 处理业务逻辑
                redisUtils.zsetRemove(DELAY_TASK_KEY, id.getValue());
            }
        }
    }
 
    @Override
    public void afterPropertiesSet() {
        new Thread(() -> {
            while (true) {
                try {
                    //5秒轮询一次，延时任务的时间误差在5秒以内
                    Thread.sleep(5 * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                consuming();
            }
        }).start();
    }
}