使用Redis實(shí)現(xiàn)延時(shí)任務(wù)的解決方案

public class DelayQueueMain {
  private static final Logger LOGGER = LoggerFactory.getLogger(DelayQueueMain.class);
  public static void main(String[] args) throws Exception {
    DelayQueueOrderMessage> queue = new DelayQueue>();
    // 默認(rèn)延遲5秒
    OrderMessage message = new OrderMessage("ORDER_ID_10086");
    queue.add(message);
    // 延遲6秒
    message = new OrderMessage("ORDER_ID_10087", 6);
    queue.add(message);
    // 延遲10秒
    message = new OrderMessage("ORDER_ID_10088", 10);
    queue.add(message);
    ExecutorService executorService = Executors.newSingleThreadExecutor(r -> {
      Thread thread = new Thread(r);
      thread.setName("DelayWorker");
      thread.setDaemon(true);
      return thread;
    });
    LOGGER.info("開始執(zhí)行調(diào)度線程...");
    executorService.execute(() -> {
      while (true) {
        try {
          OrderMessage task = queue.take();
          LOGGER.info("延遲處理訂單消息,{}", task.getDescription());
        } catch (Exception e) {
          LOGGER.error(e.getMessage(), e);
        }
      }
    });
    Thread.sleep(Integer.MAX_VALUE);
  }
  private static class OrderMessage implements Delayed {
    private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    /**
     * 默認(rèn)延遲5000毫秒
     */
    private static final long DELAY_MS = 1000L * 5;
    /**
     * 訂單ID
     */
    private final String orderId;
    /**
     * 創(chuàng)建時(shí)間戳
     */
    private final long timest
    /**
     * 過期時(shí)間
     */
    private final long expire;
    /**
     * 描述
     */
    private final String description;
    public OrderMessage(String orderId, long expireSeconds) {
      this.orderId = orderId;
      this.timestamp = System.currentTimeMillis();
      this.expire = this.timestamp + expireSeconds * 1000L;
      this.description = String.format("訂單[%s]-創(chuàng)建時(shí)間為:%s,超時(shí)時(shí)間為:%s", orderId,
          LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(F),
          LocalDateTime.ofInstant(Instant.ofEpochMilli(expire), ZoneId.systemDefault()).format(F));
    }
    public OrderMessage(String orderId) {
      this.orderId = orderId;
      this.timestamp = System.currentTimeMillis();
      this.expire = this.timestamp + DELAY_MS;
      this.description = String.format("訂單[%s]-創(chuàng)建時(shí)間為:%s,超時(shí)時(shí)間為:%s", orderId,
          LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(F),
          LocalDateTime.ofInstant(Instant.ofEpochMilli(expire), ZoneId.systemDefault()).format(F));
    }
    public String getOrderId() {
      return orderId;
    }
    public long getTimestamp() {
      return timest
    }
    public long getExpire() {
      return expire;
    }
    public String getDescription() {
      return description;
    }
    @Override
    public long getDelay(TimeUnit unit) {
      return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    @Override
    public int compareTo(Delayed o) {
      return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }
  }
}

10:16:08.240 [main] INFO club.throwable.delay.DelayQueueMain - 開始執(zhí)行調(diào)度線程...
10:16:13.224 [DelayWorker] INFO club.throwable.delay.DelayQueueMain - 延遲處理訂單消息,訂單[ORDER_ID_10086]-創(chuàng)建時(shí)間為:2019-08-20 10:16:08,超時(shí)時(shí)間為:2019-08-20 10:16:13
10:16:14.237 [DelayWorker] INFO club.throwable.delay.DelayQueueMain - 延遲處理訂單消息,訂單[ORDER_ID_10087]-創(chuàng)建時(shí)間為:2019-08-20 10:16:08,超時(shí)時(shí)間為:2019-08-20 10:16:14
10:16:18.237 [DelayWorker] INFO club.throwable.delay.DelayQueueMain - 延遲處理訂單消息,訂單[ORDER_ID_10088]-創(chuàng)建時(shí)間為:2019-08-20 10:16:08,超時(shí)時(shí)間為:2019-08-20 10:16:18

dependency>
  groupId>mysql/groupId>
  artifactId>mysql-connector-java/artifactId>
  version>5.1.48/version>
  scope>test/scope>
/dependency>
dependency>
  groupId>org.springframework.boot/groupId>
  artifactId>spring-boot-starter-jdbc/artifactId>
  version>2.1.7.RELEASE/version>
  scope>test/scope>
/dependency>
dependency>
  groupId>org.quartz-scheduler/groupId>
  artifactId>quartz/artifactId>
  version>2.3.1/version>
  scope>test/scope>
/dependency>

CREATE DATABASE `delayTask` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_520_ci;

USE `delayTask`;

CREATE TABLE `t_order_message`
(
  id      BIGINT UNSIGNED PRIMARY KEY AUTO_INCREMENT,
  order_id   VARCHAR(50) NOT NULL COMMENT '訂單ID',
  create_time DATETIME  NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創(chuàng)建日期時(shí)間',
  edit_time  DATETIME  NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '修改日期時(shí)間',
  retry_times TINYINT   NOT NULL DEFAULT 0 COMMENT '重試次數(shù)',
  order_status TINYINT   NOT NULL DEFAULT 0 COMMENT '訂單狀態(tài)',
  INDEX idx_order_id (order_id),
  INDEX idx_create_time (create_time)
) COMMENT '訂單信息表';

# 寫入兩條測(cè)試數(shù)據(jù)
INSERT INTO t_order_message(order_id) VALUES ('10086'),('10087');

// 常量
public class OrderConstants {

  public static final int MAX_RETRY_TIMES = 5;

  public static final int PENDING = 0;

  public static final int SUCCESS = 1;

  public static final int FAIL = -1;

  public static final int LIMIT = 10;
}

// 實(shí)體
@Builder
@Data
public class OrderMessage {

  private Long id;
  private String orderId;
  private LocalDateTime createTime;
  private LocalDateTime editTime;
  private Integer retryTimes;
  private Integer orderStatus;
}

// DAO
@RequiredArgsConstructor
public class OrderMessageDao {

  private final JdbcTemplate jdbcTemplate;

  private static final ResultSetExtractorListOrderMessage>> M = r -> {
    ListOrderMessage> list = Lists.newArrayList();
    while (r.next()) {
      list.add(OrderMessage.builder()
          .id(r.getLong("id"))
          .orderId(r.getString("order_id"))
          .createTime(r.getTimestamp("create_time").toLocalDateTime())
          .editTime(r.getTimestamp("edit_time").toLocalDateTime())
          .retryTimes(r.getInt("retry_times"))
          .orderStatus(r.getInt("order_status"))
          .build());
    }
    return list;
  };

  public ListOrderMessage> selectPendingRecords(LocalDateTime start,
                          LocalDateTime end,
                          ListInteger> statusList,
                          int maxRetryTimes,
                          int limit) {
    StringJoiner joiner = new StringJoiner(",");
    statusList.forEach(s -> joiner.add(String.valueOf(s)));
    return jdbcTemplate.query("SELECT * FROM t_order_message WHERE create_time >= ? AND create_time = ? " +
            "AND order_status IN (?) AND retry_times  ? LIMIT ?",
        p -> {
          p.setTimestamp(1, Timestamp.valueOf(start));
          p.setTimestamp(2, Timestamp.valueOf(end));
          p.setString(3, joiner.toString());
          p.setInt(4, maxRetryTimes);
          p.setInt(5, limit);
        }, M);
  }

  public int updateOrderStatus(Long id, int status) {
    return jdbcTemplate.update("UPDATE t_order_message SET order_status = ?,edit_time = ? WHERE id =?",
        p -> {
          p.setInt(1, status);
          p.setTimestamp(2, Timestamp.valueOf(LocalDateTime.now()));
          p.setLong(3, id);
        });
  }
}

// Service
@RequiredArgsConstructor
public class OrderMessageService {

  private static final Logger LOGGER = LoggerFactory.getLogger(OrderMessageService.class);

  private final OrderMessageDao orderMessageDao;

  private static final ListInteger> STATUS = Lists.newArrayList();

  static {
    STATUS.add(OrderConstants.PENDING);
    STATUS.add(OrderConstants.FAIL);
  }

  public void executeDelayJob() {
    LOGGER.info("訂單處理定時(shí)任務(wù)開始執(zhí)行......");
    LocalDateTime end = LocalDateTime.now();
    // 一天前
    LocalDateTime start = end.minusDays(1);
    ListOrderMessage> list = orderMessageDao.selectPendingRecords(start, end, STATUS, OrderConstants.MAX_RETRY_TIMES, OrderConstants.LIMIT);
    if (!list.isEmpty()) {
      for (OrderMessage m : list) {
        LOGGER.info("處理訂單[{}],狀態(tài)由{}更新為{}", m.getOrderId(), m.getOrderStatus(), OrderConstants.SUCCESS);
        // 這里其實(shí)可以優(yōu)化為批量更新
        orderMessageDao.updateOrderStatus(m.getId(), OrderConstants.SUCCESS);
      }
    }
    LOGGER.info("訂單處理定時(shí)任務(wù)開始完畢......");
  }
}

// Job
@DisallowConcurrentExecution
public class OrderMessageDelayJob implements Job {

  @Override
  public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
    OrderMessageService service = (OrderMessageService) jobExecutionContext.getMergedJobDataMap().get("orderMessageService");
    service.executeDelayJob();
  }

  public static void main(String[] args) throws Exception {
    HikariConfig config = new HikariConfig();
    config.setJdbcUrl("jdbc:mysql://localhost:3306/delayTask?useSSL=falsecharacterEncoding=utf8");
    config.setDriverClassName(Driver.class.getName());
    config.setUsername("root");
    config.setPassword("root");
    HikariDataSource dataSource = new HikariDataSource(config);
    OrderMessageDao orderMessageDao = new OrderMessageDao(new JdbcTemplate(dataSource));
    OrderMessageService service = new OrderMessageService(orderMessageDao);
    // 內(nèi)存模式的調(diào)度器
    StdSchedulerFactory factory = new StdSchedulerFactory();
    Scheduler scheduler = factory.getScheduler();
    // 這里沒有用到IOC容器,直接用Quartz數(shù)據(jù)集合傳遞服務(wù)引用
    JobDataMap jobDataMap = new JobDataMap();
    jobDataMap.put("orderMessageService", service);
    // 新建Job
    JobDetail job = JobBuilder.newJob(OrderMessageDelayJob.class)
        .withIdentity("orderMessageDelayJob", "delayJob")
        .usingJobData(jobDataMap)
        .build();
    // 新建觸發(fā)器,10秒執(zhí)行一次
    Trigger trigger = TriggerBuilder.newTrigger()
        .withIdentity("orderMessageDelayTrigger", "delayJob")
        .withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(10).repeatForever())
        .build();
    scheduler.scheduleJob(job, trigger);
    // 啟動(dòng)調(diào)度器
    scheduler.start();
    Thread.sleep(Integer.MAX_VALUE);
  }
}

11:58:27.202 [main] INFO org.quartz.core.QuartzScheduler - Scheduler meta-data: Quartz Scheduler (v2.3.1) 'DefaultQuartzScheduler' with instanceId 'NON_CLUSTERED'
 Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.
 NOT STARTED.
 Currently in standby mode.
 Number of jobs executed: 0
 Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 10 threads.
 Using job-store 'org.quartz.simpl.RAMJobStore' - which does not support persistence. and is not clustered.

11:58:27.202 [main] INFO org.quartz.impl.StdSchedulerFactory - Quartz scheduler 'DefaultQuartzScheduler' initialized from default resource file in Quartz package: 'quartz.properties'
11:58:27.202 [main] INFO org.quartz.impl.StdSchedulerFactory - Quartz scheduler version: 2.3.1
11:58:27.209 [main] INFO org.quartz.core.QuartzScheduler - Scheduler DefaultQuartzScheduler_$_NON_CLUSTERED started.
11:58:27.212 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 1 triggers
11:58:27.217 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.simpl.PropertySettingJobFactory - Producing instance of Job 'delayJob.orderMessageDelayJob', class=club.throwable.jdbc.OrderMessageDelayJob
11:58:27.219 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@10eb8c53
11:58:27.220 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 0 triggers
11:58:27.221 [DefaultQuartzScheduler_Worker-1] DEBUG org.quartz.core.JobRunShell - Calling execute on job delayJob.orderMessageDelayJob
11:58:34.440 [DefaultQuartzScheduler_Worker-1] INFO club.throwable.jdbc.OrderMessageService - 訂單處理定時(shí)任務(wù)開始執(zhí)行......
11:58:34.451 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@3d27ece4
11:58:34.459 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@64e808af
11:58:34.470 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@79c8c2b7
11:58:34.477 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@19a62369
11:58:34.485 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@1673d017
11:58:34.485 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - After adding stats (total=10, active=0, idle=10, waiting=0)
11:58:34.559 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL query
11:58:34.565 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL statement [SELECT * FROM t_order_message WHERE create_time >= ? AND create_time = ? AND order_status IN (?) AND retry_times  ? LIMIT ?]
11:58:34.645 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.datasource.DataSourceUtils - Fetching JDBC Connection from DataSource
11:58:35.210 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - SQLWarning ignored: SQL state '22007', error code '1292', message [Truncated incorrect DOUBLE value: '0,-1']
11:58:35.335 [DefaultQuartzScheduler_Worker-1] INFO club.throwable.jdbc.OrderMessageService - 處理訂單[10086],狀態(tài)由0更新為1
11:58:35.342 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL update
11:58:35.346 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL statement [UPDATE t_order_message SET order_status = ?,edit_time = ? WHERE id =?]
11:58:35.347 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.datasource.DataSourceUtils - Fetching JDBC Connection from DataSource
11:58:35.354 [DefaultQuartzScheduler_Worker-1] INFO club.throwable.jdbc.OrderMessageService - 處理訂單[10087],狀態(tài)由0更新為1
11:58:35.355 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL update
11:58:35.355 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL statement [UPDATE t_order_message SET order_status = ?,edit_time = ? WHERE id =?]
11:58:35.355 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.datasource.DataSourceUtils - Fetching JDBC Connection from DataSource
11:58:35.361 [DefaultQuartzScheduler_Worker-1] INFO club.throwable.jdbc.OrderMessageService - 訂單處理定時(shí)任務(wù)開始完畢......
11:58:35.363 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 1 triggers
11:58:37.206 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.simpl.PropertySettingJobFactory - Producing instance of Job 'delayJob.orderMessageDelayJob', class=club.throwable.jdbc.OrderMessageDelayJob
11:58:37.206 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 0 triggers

dependency>
  groupId>com.rabbitmq/groupId>
  artifactId>amqp-client/artifactId>
  version>5.7.3/version>
  scope>test/scope>
/dependency>

public class DlxMain {

  private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  private static final Logger LOGGER = LoggerFactory.getLogger(DlxMain.class);

  public static void main(String[] args) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    Connection connection = factory.newConnection();
    Channel producerChannel = connection.createChannel();
    Channel consumerChannel = connection.createChannel();
    // dlx交換器名稱為dlx.exchange,類型是direct,綁定鍵為dlx.key,隊(duì)列名為dlx.queue
    producerChannel.exchangeDeclare("dlx.exchange", "direct");
    producerChannel.queueDeclare("dlx.queue", false, false, false, null);
    producerChannel.queueBind("dlx.queue", "dlx.exchange", "dlx.key");
    MapString, Object> queueArgs = new HashMap>();
    // 設(shè)置隊(duì)列消息過期時(shí)間,5秒
    queueArgs.put("x-message-ttl", 5000);
    // 指定DLX相關(guān)參數(shù)
    queueArgs.put("x-dead-letter-exchange", "dlx.exchange");
    queueArgs.put("x-dead-letter-routing-key", "dlx.key");
    // 聲明業(yè)務(wù)隊(duì)列
    producerChannel.queueDeclare("business.queue", false, false, false, queueArgs);
    ExecutorService executorService = Executors.newSingleThreadExecutor(r -> {
      Thread thread = new Thread(r);
      thread.setDaemon(true);
      thread.setName("DlxConsumer");
      return thread;
    });
    // 啟動(dòng)消費(fèi)者
    executorService.execute(() -> {
      try {
        consumerChannel.basicConsume("dlx.queue", true, new DlxConsumer(consumerChannel));
      } catch (IOException e) {
        LOGGER.error(e.getMessage(), e);
      }
    });
    OrderMessage message = new OrderMessage("10086");
    producerChannel.basicPublish("", "business.queue", MessageProperties.TEXT_PLAIN,
        message.getDescription().getBytes(StandardCharsets.UTF_8));
    LOGGER.info("發(fā)送消息成功,訂單ID:{}", message.getOrderId());

    message = new OrderMessage("10087");
    producerChannel.basicPublish("", "business.queue", MessageProperties.TEXT_PLAIN,
        message.getDescription().getBytes(StandardCharsets.UTF_8));
    LOGGER.info("發(fā)送消息成功,訂單ID:{}", message.getOrderId());

    message = new OrderMessage("10088");
    producerChannel.basicPublish("", "business.queue", MessageProperties.TEXT_PLAIN,
        message.getDescription().getBytes(StandardCharsets.UTF_8));
    LOGGER.info("發(fā)送消息成功,訂單ID:{}", message.getOrderId());

    Thread.sleep(Integer.MAX_VALUE);
  }

  private static class DlxConsumer extends DefaultConsumer {

    DlxConsumer(Channel channel) {
      super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag,
                  Envelope envelope,
                  AMQP.BasicProperties properties,
                  byte[] body) throws IOException {
      LOGGER.info("處理消息成功:{}", new String(body, StandardCharsets.UTF_8));
    }
  }

  private static class OrderMessage {

    private final String orderId;
    private final long timest
    private final String description;

    OrderMessage(String orderId) {
      this.orderId = orderId;
      this.timestamp = System.currentTimeMillis();
      this.description = String.format("訂單[%s],訂單創(chuàng)建時(shí)間為:%s", orderId,
          LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(F));
    }

    public String getOrderId() {
      return orderId;
    }

    public long getTimestamp() {
      return timest
    }

    public String getDescription() {
      return description;
    }
  }
}

16:35:58.638 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:10086
16:35:58.641 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:10087
16:35:58.641 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:10088
16:36:03.646 [pool-1-thread-4] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10086],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:58
16:36:03.670 [pool-1-thread-5] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10087],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:58
16:36:03.670 [pool-1-thread-6] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10088],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:58

dependency>
  groupId>io.netty/groupId>
  artifactId>netty-common/artifactId>
  version>4.1.39.Final/version>
/dependency>

public class HashedWheelTimerMain {

  private static final DateTimeFormatter F = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");

  public static void main(String[] args) throws Exception {
    AtomicInteger counter = new AtomicInteger();
    ThreadFactory factory = r -> {
      Thread thread = new Thread(r);
      thread.setDaemon(true);
      thread.setName("HashedWheelTimerWorker-" + counter.getAndIncrement());
      return thread;
    };
    // tickDuration - 每tick一次的時(shí)間間隔, 每tick一次就會(huì)到達(dá)下一個(gè)槽位
    // unit - tickDuration的時(shí)間單位
    // ticksPerWhee - 時(shí)間輪中的槽位數(shù)
    Timer timer = new HashedWheelTimer(factory, 1, TimeUnit.SECONDS, 60);
    TimerTask timerTask = new DefaultTimerTask("10086");
    timer.newTimeout(timerTask, 5, TimeUnit.SECONDS);
    timerTask = new DefaultTimerTask("10087");
    timer.newTimeout(timerTask, 10, TimeUnit.SECONDS);
    timerTask = new DefaultTimerTask("10088");
    timer.newTimeout(timerTask, 15, TimeUnit.SECONDS);
    Thread.sleep(Integer.MAX_VALUE);
  }

  private static class DefaultTimerTask implements TimerTask {

    private final String orderId;
    private final long timest

    public DefaultTimerTask(String orderId) {
      this.orderId = orderId;
      this.timestamp = System.currentTimeMillis();
    }

    @Override
    public void run(Timeout timeout) throws Exception {
      System.out.println(String.format("任務(wù)執(zhí)行時(shí)間:%s,訂單創(chuàng)建時(shí)間:%s,訂單ID:%s",
          LocalDateTime.now().format(F), LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()).format(F), orderId));
    }
  }
}

dependencyManagement>
  dependencies>
    dependency>
      groupId>org.springframework.boot/groupId>
      artifactId>spring-boot-dependencies/artifactId>
      version>2.1.7.RELEASE/version>
      type>pom/type>
      scope>import/scope>
    /dependency>
  /dependencies>
/dependencyManagement>

dependencies>
  dependency>
    groupId>org.quartz-scheduler/groupId>
    artifactId>quartz/artifactId>
    version>2.3.1/version>
  /dependency>
  dependency>
    groupId>redis.clients/groupId>
    artifactId>jedis/artifactId>
    version>3.1.0/version>
  /dependency>
  dependency>
    groupId>org.springframework.boot/groupId>
    artifactId>spring-boot-starter-web/artifactId>
  /dependency>
  dependency>
    groupId>org.springframework.boot/groupId>
    artifactId>spring-boot-starter-jdbc/artifactId>
  /dependency>  
  dependency>
    groupId>org.springframework/groupId>
    artifactId>spring-context-support/artifactId>
    version>5.1.9.RELEASE/version>
  /dependency> 
  dependency>
    groupId>org.projectlombok/groupId>
    artifactId>lombok/artifactId>
    version>1.18.8/version>
    scope>provided/scope>
  /dependency>
  dependency>
    groupId>com.alibaba/groupId>
    artifactId>fastjson/artifactId>
    version>1.2.59/version>
  /dependency>    
/dependencies>

-- /lua/enqueue.lua
local zset_key = KEYS[1]
local hash_key = KEYS[2]
local zset_value = ARGV[1]
local zset_score = ARGV[2]
local hash_field = ARGV[3]
local hash_value = ARGV[4]
redis.call('ZADD', zset_key, zset_score, zset_value)
redis.call('HSET', hash_key, hash_field, hash_value)
return nil

-- /lua/dequeue.lua
-- 參考jesque的部分Lua腳本實(shí)現(xiàn)
local zset_key = KEYS[1]
local hash_key = KEYS[2]
local min_score = ARGV[1]
local max_score = ARGV[2]
local offset = ARGV[3]
local limit = ARGV[4]
-- TYPE命令的返回結(jié)果是{'ok':'zset'}這樣子,這里利用next做一輪迭代
local status, type = next(redis.call('TYPE', zset_key))
if status ~= nil and status == 'ok' then
  if type == 'zset' then
    local list = redis.call('ZREVRANGEBYSCORE', zset_key, max_score, min_score, 'LIMIT', offset, limit)
    if list ~= nil and #list > 0 then
      -- unpack函數(shù)能把table轉(zhuǎn)化為可變參數(shù)
      redis.call('ZREM', zset_key, unpack(list))
      local result = redis.call('HMGET', hash_key, unpack(list))
      redis.call('HDEL', hash_key, unpack(list))
      return result
    end
  end
end
return nil

// Jedis提供者
@Component
public class JedisProvider implements InitializingBean {

  private JedisPool jedisPool;

  @Override
  public void afterPropertiesSet() throws Exception {
    jedisPool = new JedisPool();
  }

  public Jedis provide(){
    return jedisPool.getResource();
  }
}

// OrderMessage
@Data
public class OrderMessage {

  private String orderId;
  private BigDecimal amount;
  private Long userId;
}

// 延遲隊(duì)列接口
public interface OrderDelayQueue {

  void enqueue(OrderMessage message);

  ListOrderMessage> dequeue(String min, String max, String offset, String limit);

  ListOrderMessage> dequeue();

  String enqueueSha();

  String dequeueSha();
}

// 延遲隊(duì)列實(shí)現(xiàn)類
@RequiredArgsConstructor
@Component
public class RedisOrderDelayQueue implements OrderDelayQueue, InitializingBean {

  private static final String MIN_SCORE = "0";
  private static final String OFFSET = "0";
  private static final String LIMIT = "10";
  private static final String ORDER_QUEUE = "ORDER_QUEUE";
  private static final String ORDER_DETAIL_QUEUE = "ORDER_DETAIL_QUEUE";
  private static final String ENQUEUE_LUA_SCRIPT_LOCATION = "/lua/enqueue.lua";
  private static final String DEQUEUE_LUA_SCRIPT_LOCATION = "/lua/dequeue.lua";
  private static final AtomicReferenceString> ENQUEUE_LUA_SHA = new AtomicReference>();
  private static final AtomicReferenceString> DEQUEUE_LUA_SHA = new AtomicReference>();
  private static final ListString> KEYS = Lists.newArrayList();

  private final JedisProvider jedisProvider;

  static {
    KEYS.add(ORDER_QUEUE);
    KEYS.add(ORDER_DETAIL_QUEUE);
  }

  @Override
  public void enqueue(OrderMessage message) {
    ListString> args = Lists.newArrayList();
    args.add(message.getOrderId());
    args.add(String.valueOf(System.currentTimeMillis()));
    args.add(message.getOrderId());
    args.add(JSON.toJSONString(message));
    try (Jedis jedis = jedisProvider.provide()) {
      jedis.evalsha(ENQUEUE_LUA_SHA.get(), KEYS, args);
    }
  }

  @Override
  public ListOrderMessage> dequeue() {
    // 30分鐘之前
    String maxScore = String.valueOf(System.currentTimeMillis() - 30 * 60 * 1000);
    return dequeue(MIN_SCORE, maxScore, OFFSET, LIMIT);
  }

  @SuppressWarnings("unchecked")
  @Override
  public ListOrderMessage> dequeue(String min, String max, String offset, String limit) {
    ListString> args = new ArrayList>();
    args.add(max);
    args.add(min);
    args.add(offset);
    args.add(limit);
    ListOrderMessage> result = Lists.newArrayList();
    try (Jedis jedis = jedisProvider.provide()) {
      ListString> eval = (ListString>) jedis.evalsha(DEQUEUE_LUA_SHA.get(), KEYS, args);
      if (null != eval) {
        for (String e : eval) {
          result.add(JSON.parseObject(e, OrderMessage.class));
        }
      }
    }
    return result;
  }

  @Override
  public String enqueueSha() {
    return ENQUEUE_LUA_SHA.get();
  }

  @Override
  public String dequeueSha() {
    return DEQUEUE_LUA_SHA.get();
  }

  @Override
  public void afterPropertiesSet() throws Exception {
    // 加載Lua腳本
    loadLuaScript();
  }

  private void loadLuaScript() throws Exception {
    try (Jedis jedis = jedisProvider.provide()) {
      ClassPathResource resource = new ClassPathResource(ENQUEUE_LUA_SCRIPT_LOCATION);
      String luaContent = StreamUtils.copyToString(resource.getInputStream(), StandardCharsets.UTF_8);
      String sha = jedis.scriptLoad(luaContent);
      ENQUEUE_LUA_SHA.compareAndSet(null, sha);
      resource = new ClassPathResource(DEQUEUE_LUA_SCRIPT_LOCATION);
      luaContent = StreamUtils.copyToString(resource.getInputStream(), StandardCharsets.UTF_8);
      sha = jedis.scriptLoad(luaContent);
      DEQUEUE_LUA_SHA.compareAndSet(null, sha);
    }
  }

  public static void main(String[] as) throws Exception {
    DateTimeFormatter f = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
    JedisProvider jedisProvider = new JedisProvider();
    jedisProvider.afterPropertiesSet();
    RedisOrderDelayQueue queue = new RedisOrderDelayQueue(jedisProvider);
    queue.afterPropertiesSet();
    // 寫入測(cè)試數(shù)據(jù)
    OrderMessage message = new OrderMessage();
    message.setAmount(BigDecimal.valueOf(10086));
    message.setOrderId("ORDER_ID_10086");
    message.setUserId(10086L);
    message.setTimestamp(LocalDateTime.now().format(f));
    ListString> args = Lists.newArrayList();
    args.add(message.getOrderId());
    // 測(cè)試需要,score設(shè)置為30分鐘之前
    args.add(String.valueOf(System.currentTimeMillis() - 30 * 60 * 1000));
    args.add(message.getOrderId());
    args.add(JSON.toJSONString(message));
    try (Jedis jedis = jedisProvider.provide()) {
      jedis.evalsha(ENQUEUE_LUA_SHA.get(), KEYS, args);
    }
    ListOrderMessage> dequeue = queue.dequeue();
    System.out.println(dequeue);
  }
}

[OrderMessage(orderId=ORDER_ID_10086, amount=10086, userId=10086, timestamp=2019-08-21 08:32:22.885)]

@DisallowConcurrentExecution
@Component
public class OrderMessageConsumer implements Job {

  private static final AtomicInteger COUNTER = new AtomicInteger();
  private static final ExecutorService BUSINESS_WORKER_POOL = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), r -> {
    Thread thread = new Thread(r);
    thread.setDaemon(true);
    thread.setName("OrderMessageConsumerWorker-" + COUNTER.getAndIncrement());
    return thread;
  });
  private static final Logger LOGGER = LoggerFactory.getLogger(OrderMessageConsumer.class);

  @Autowired
  private OrderDelayQueue orderDelayQueue;

  @Override
  public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();
    LOGGER.info("訂單消息處理定時(shí)任務(wù)開始執(zhí)行......");
    ListOrderMessage> messages = orderDelayQueue.dequeue();
    if (!messages.isEmpty()) {
      // 簡單的列表等分放到線程池中執(zhí)行
      ListListOrderMessage>> partition = Lists.partition(messages, 2);
      int size = partition.size();
      final CountDownLatch latch = new CountDownLatch(size);
      for (ListOrderMessage> p : partition) {
        BUSINESS_WORKER_POOL.execute(new ConsumeTask(p, latch));
      }
      try {
        latch.await();
      } catch (InterruptedException ignore) {
        //ignore
      }
    }
    stopWatch.stop();
    LOGGER.info("訂單消息處理定時(shí)任務(wù)執(zhí)行完畢,耗時(shí):{} ms......", stopWatch.getTotalTimeMillis());
  }

  @RequiredArgsConstructor
  private static class ConsumeTask implements Runnable {

    private final ListOrderMessage> messages;
    private final CountDownLatch latch;

    @Override
    public void run() {
      try {
        // 實(shí)際上這里應(yīng)該單條捕獲異常
        for (OrderMessage message : messages) {
          LOGGER.info("處理訂單信息,內(nèi)容:{}", message);
        }
      } finally {
        latch.countDown();
      }
    }
  }
}

// Quartz配置類
@Configuration
public class QuartzAutoConfiguration {

  @Bean
  public SchedulerFactoryBean schedulerFactoryBean(QuartzAutowiredJobFactory quartzAutowiredJobFactory) {
    SchedulerFactoryBean factory = new SchedulerFactoryBean();
    factory.setAutoStartup(true);
    factory.setJobFactory(quartzAutowiredJobFactory);
    return factory;
  }

  @Bean
  public QuartzAutowiredJobFactory quartzAutowiredJobFactory() {
    return new QuartzAutowiredJobFactory();
  }

  public static class QuartzAutowiredJobFactory extends AdaptableJobFactory implements BeanFactoryAware {

    private AutowireCapableBeanFactory autowireCapableBeanFactory;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
      this.autowireCapableBeanFactory = (AutowireCapableBeanFactory) beanFactory;
    }

    @Override
    protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
      Object jobInstance = super.createJobInstance(bundle);
      // 這里利用AutowireCapableBeanFactory從新建的Job實(shí)例做一次自動(dòng)裝配,得到一個(gè)原型(prototype)的JobBean實(shí)例
      autowireCapableBeanFactory.autowireBean(jobInstance);
      return jobInstance;
    }
  }
}

@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, TransactionAutoConfiguration.class})
public class Application implements CommandLineRunner {

  @Autowired
  private Scheduler scheduler;

  @Autowired
  private JedisProvider jedisProvider;

  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }

  @Override
  public void run(String... args) throws Exception {
    // 準(zhǔn)備一些測(cè)試數(shù)據(jù)
    prepareOrderMessageData();
    JobDetail job = JobBuilder.newJob(OrderMessageConsumer.class)
        .withIdentity("OrderMessageConsumer", "DelayTask")
        .build();
    // 觸發(fā)器5秒觸發(fā)一次
    Trigger trigger = TriggerBuilder.newTrigger()
        .withIdentity("OrderMessageConsumerTrigger", "DelayTask")
        .withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(5).repeatForever())
        .build();
    scheduler.scheduleJob(job, trigger);
  }

  private void prepareOrderMessageData() throws Exception {
    DateTimeFormatter f = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
    try (Jedis jedis = jedisProvider.provide()) {
      ListOrderMessage> messages = Lists.newArrayList();
      for (int i = 0; i  100; i++) {
        OrderMessage message = new OrderMessage();
        message.setAmount(BigDecimal.valueOf(i));
        message.setOrderId("ORDER_ID_" + i);
        message.setUserId((long) i);
        message.setTimestamp(LocalDateTime.now().format(f));
        messages.add(message);
      }
      // 這里暫時(shí)不使用Lua
      MapString, Double> map = Maps.newHashMap();
      MapString, String> hash = Maps.newHashMap();
      for (OrderMessage message : messages) {
        // 故意把score設(shè)計(jì)成30分鐘前
        map.put(message.getOrderId(), Double.valueOf(String.valueOf(System.currentTimeMillis() - 30 * 60 * 1000)));
        hash.put(message.getOrderId(), JSON.toJSONString(message));
      }
      jedis.zadd("ORDER_QUEUE", map);
      jedis.hmset("ORDER_DETAIL_QUEUE", hash);
    }
  }
}