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

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

一、介绍

本文主要介绍如何使用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);
                  }
              });
          }
      }
  }
  

三、运行效果

四、源码

https://github.com/1277463718lmt/spring-plugin-demo.git