使用spring-plugin和redisson实现延迟队列-526互联

一、介绍
二、步骤
三、运行效果
四、源码

一、介绍

本文主要介绍如何使用spring plugin和redisson去实现延迟队列

二、步骤

pom.xml引入依赖包

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.plugin</groupId>
            <artifactId>spring-plugin-core</artifactId>
            <version>2.0.0.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.16.3</version>
        </dependency>
    </dependencies>

配置文件application.properties

server.port=18084

spring.redis.host=192.168.48.48
spring.redis.port=6379
spring.redis.database=1
spring.redis.password=

定义延迟队列任务的枚举类DelayQueueTypes.java

public enum DelayQueueTypes {
    TASK_1("task-1"),
    TASK_2("task-2");

    public String name;

    DelayQueueTypes(String name) {
        this.name = name;
    }
}

定义延迟队列的插件DelayQueueHandler.java

public interface DelayQueueHandler extends Plugin<DelayQueueTypes> {
    void execute(Object data);
}

实现不同任务的插件

@Component
@Order(value = 1)
public class Task1DelayQueueTypesHandler implements DelayQueueHandler {
    private static final Logger log = LoggerFactory.getLogger(Task1DelayQueueTypesHandler.class);

    @Override
    public void execute(Object data) {
        log.info("任务一接收到数据 = {}", data);
    }

    @Override
    public boolean supports(DelayQueueTypes delayQueueType) {
        return DelayQueueTypes.TASK_1 == delayQueueType;
    }
}

@Component
@Order(value = -2)
public class Task2DelayQueueTypesHandler implements DelayQueueHandler {
    private static final Logger log = LoggerFactory.getLogger(Task1DelayQueueTypesHandler.class);

    @Override
    public void execute(Object data) {
        log.info("任务二接收到数据 = {}", data);
    }

    @Override
    public boolean supports(DelayQueueTypes delayQueueType) {
        return DelayQueueTypes.TASK_2 == delayQueueType;
    }
}

注册插件

@Configuration
@EnablePluginRegistries(value = {DelayQueueHandler.class})
public class PluginConfiguration {
}

定义Redis配置类RedisConfig.java

@Configuration
public class RedisConfig {

    @Bean
    public RedissonClient redissonClient(RedisProperties redisProperties) {
        Config config = new Config();
        SingleServerConfig singleServerConfig = config.useSingleServer().setAddress("redis://" + redisProperties.getHost() + ":" + redisProperties.getPort());
        if (!StringUtils.isEmpty(redisProperties.getPassword())) {
            singleServerConfig.setPassword(redisProperties.getPassword());
        }
        return Redisson.create(config);
    }
}

定义延迟队列的工具类RedisDelayQueueUtils.java


@Component
public class RedisDelayQueueUtils {
    private static final Logger logger = LoggerFactory.getLogger(RedisDelayQueueUtils.class);

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 加入队列
     *
     * @param t
     * @param l
     * @param unit
     * @param delayQueueType
     * @param <T>
     * @return
     */
    public <T> int addDelayQueue(T t, long l, TimeUnit unit, DelayQueueTypes delayQueueType) {
        try {
            RBlockingQueue<T> rBlockingQueue = redissonClient.getBlockingQueue(delayQueueType.name);
            RDelayedQueue<T> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
            rDelayedQueue.offer(t, l, unit);
            logger.info("[{}队列]增加元素[{}], 有效期：{} {}", delayQueueType.name, t.toString(), l, unit.toString());
            return 1;
        } catch (Exception e) {
            logger.error("[{}队列]增加元素失败", delayQueueType.name);
            return -1;
        }
    }

    /**
     * 加入/替换队列（保证唯一）
     *
     * @param t
     * @param l
     * @param unit
     * @param delayQueueType
     * @param <T>
     * @return
     */
    public <T> int addOrUpdateDelayQueue(T t, long l, TimeUnit unit, DelayQueueTypes delayQueueType) {
        try {
            RBlockingQueue<T> rBlockingQueue = redissonClient.getBlockingQueue(delayQueueType.name);
            RDelayedQueue<T> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
            // 先清空再加入
            rDelayedQueue.removeAll(Arrays.asList(t));
            rDelayedQueue.offer(t, l, unit);
            logger.info("[{}队列]增加元素[{}], 有效期：{}", delayQueueType.name, t.toString(), l);
            return 1;
        } catch (Exception e) {
            logger.error("[{}队列]增加元素失败", delayQueueType.name);
            return -1;
        }
    }

    /**
     * 获取队列
     *
     * @param delayQueueType
     * @return
     * @throws InterruptedException
     */
    public <T> T getDelayQueue(DelayQueueTypes delayQueueType) throws InterruptedException {
        if (redissonClient != null) {
            RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(delayQueueType.name);
            redissonClient.getDelayedQueue(blockingDeque);
            return blockingDeque.take();
        }
        return null;
    }
}

定义延迟队列的启动运行器DelayQueueHandlerRunner.java

@Component
public class DelayQueueHandlerRunner implements CommandLineRunner {
    private ExecutorService delayQueueHandleThreadPool = new ThreadPoolExecutor(5, 20, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

    @Autowired
    private PluginRegistry<DelayQueueHandler, DelayQueueTypes> delayQueueTypesPluginRegistry;

    @Autowired
    private RedisDelayQueueUtils redisDelayQueueUtils;

    @Override
    public void run(String... args) throws Exception {
        // 这边添加数据到延迟队列主要是为了测试
        redisDelayQueueUtils.addDelayQueue("这是任务一的数据", 5, TimeUnit.SECONDS, DelayQueueTypes.TASK_1);
        redisDelayQueueUtils.addDelayQueue("这是任务一的数据", 10, TimeUnit.SECONDS, DelayQueueTypes.TASK_1);
        redisDelayQueueUtils.addDelayQueue("这是任务二的数据", 15, TimeUnit.SECONDS, DelayQueueTypes.TASK_2);

        if (delayQueueTypesPluginRegistry == null) return;

        DelayQueueTypes[] delayQueueTypes = DelayQueueTypes.values();
        for (DelayQueueTypes delayQueueType : delayQueueTypes) {
            delayQueueHandleThreadPool.submit(() -> {
                Object data = null;
                try {
                    data = redisDelayQueueUtils.getDelayQueue(delayQueueType);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                List<DelayQueueHandler> delayQueueHandlerList = delayQueueTypesPluginRegistry.getPluginsFor(delayQueueType);
                for (DelayQueueHandler delayQueueHandler : delayQueueHandlerList) {
                    delayQueueHandler.execute(data);
                }
            });
        }
    }
}