最近在生產(chǎn)環(huán)境剛好遇到了延時(shí)任務(wù)的場景，調(diào)研了一下目前主流的方案，分析了一下優(yōu)劣并且敲定了最終的方案。這篇文章記錄了調(diào)研的過程，以及初步方案的實(shí)現(xiàn)。
候選方案對(duì)比
下面是想到的幾種實(shí)現(xiàn)延時(shí)任務(wù)的方案，總結(jié)了一下相應(yīng)的優(yōu)勢和劣勢。
| 方案 | 優(yōu)勢 | 劣勢 | 選用場景 |
|---|---|---|---|
JDK 內(nèi)置的延遲隊(duì)列 DelayQueue | 實(shí)現(xiàn)簡單 | 數(shù)據(jù)內(nèi)存態(tài)，不可靠 | 一致性相對(duì)低的場景 |
調(diào)度框架和 MySQL 進(jìn)行短間隔輪詢 | 實(shí)現(xiàn)簡單，可靠性高 | 存在明顯的性能瓶頸 | 數(shù)據(jù)量較少實(shí)時(shí)性相對(duì)低的場景 |
RabbitMQ 的 DLX 和 TTL，一般稱為 死信隊(duì)列 方案 | 異步交互可以削峰 | 延時(shí)的時(shí)間長度不可控，如果數(shù)據(jù)需要持久化則性能會(huì)降低 | - |
調(diào)度框架和 Redis 進(jìn)行短間隔輪詢 | 數(shù)據(jù)持久化，高性能 | 實(shí)現(xiàn)難度大 | 常見于支付結(jié)果回調(diào)方案 |
| 時(shí)間輪 | 實(shí)時(shí)性高 | 實(shí)現(xiàn)難度大，內(nèi)存消耗大 | 實(shí)時(shí)性高的場景 |
如果應(yīng)用的數(shù)據(jù)量不高，實(shí)時(shí)性要求比較低，選用調(diào)度框架和 MySQL 進(jìn)行短間隔輪詢這個(gè)方案是最優(yōu)的方案。但是筆者遇到的場景數(shù)據(jù)量相對(duì)比較大，實(shí)時(shí)性并不高，采用掃庫的方案一定會(huì)對(duì) MySQL 實(shí)例造成比較大的壓力。記得很早之前，看過一個(gè)PPT叫《盒子科技聚合支付系統(tǒng)演進(jìn)》，其中里面有一張圖片給予筆者一點(diǎn)啟發(fā)：
里面剛好用到了調(diào)度框架和 Redis 進(jìn)行短間隔輪詢實(shí)現(xiàn)延時(shí)任務(wù)的方案，不過為了分?jǐn)倯?yīng)用的壓力，圖中的方案還做了分片處理。鑒于筆者當(dāng)前業(yè)務(wù)緊迫，所以在第一期的方案暫時(shí)不考慮分片，只做了一個(gè)簡化版的實(shí)現(xiàn)。
由于PPT中沒有任何的代碼或者框架貼出，有些需要解決的技術(shù)點(diǎn)需要自行思考，下面會(huì)重現(xiàn)一次整個(gè)方案實(shí)現(xiàn)的詳細(xì)過程。
場景設(shè)計(jì)
實(shí)際的生產(chǎn)場景是筆者負(fù)責(zé)的某個(gè)系統(tǒng)需要對(duì)接一個(gè)外部的資金方，每一筆資金下單后需要延時(shí)30分鐘推送對(duì)應(yīng)的附件。這里簡化為一個(gè)訂單信息數(shù)據(jù)延遲處理的場景，就是每一筆下單記錄一條訂單消息(暫時(shí)叫做 OrderMessage )，訂單消息需要延遲5到15秒后進(jìn)行異步處理。
否決的候選方案實(shí)現(xiàn)思路
下面介紹一下其它四個(gè)不選用的候選方案，結(jié)合一些偽代碼和流程分析一下實(shí)現(xiàn)過程。
JDK內(nèi)置延遲隊(duì)列
DelayQueue 是一個(gè)阻塞隊(duì)列的實(shí)現(xiàn)，它的隊(duì)列元素必須是 Delayed 的子類，這里做個(gè)簡單的例子：
public class DelayQueueMain { private static final Logger LOGGER = LoggerFactory.getLogger(DelayQueueMain.class); public static void main(String[] args) throws Exception { DelayQueue<OrderMessage> 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 timestamp; /** * 過期時(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 timestamp; } 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)); } }}注意一下， OrderMessage 實(shí)現(xiàn) Delayed 接口，關(guān)鍵是需要實(shí)現(xiàn) Delayed#getDelay() 和 Delayed#compareTo() 。運(yùn)行一下 main() 方法：
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:1310: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:1410: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
調(diào)度框架 + MySQL
使用調(diào)度框架對(duì) MySQL 表進(jìn)行短間隔輪詢是實(shí)現(xiàn)難度比較低的方案，通常服務(wù)剛上線，表數(shù)據(jù)不多并且實(shí)時(shí)性不高的情況下應(yīng)該首選這個(gè)方案。不過要注意以下幾點(diǎn)：
MySQL
引入 Quartz 、 MySQL 的Java驅(qū)動(dòng)包和 spring-boot-starter-jdbc （這里只是為了方便用相對(duì)輕量級(jí)的框架實(shí)現(xiàn)，生產(chǎn)中可以按場景按需選擇其他更合理的框架）：
<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>
假設(shè)表設(shè)計(jì)如下：
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 '訂單信息表';# 寫入兩條測試數(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@Datapublic class OrderMessage { private Long id; private String orderId; private LocalDateTime createTime; private LocalDateTime editTime; private Integer retryTimes; private Integer orderStatus;}// DAO@RequiredArgsConstructorpublic class OrderMessageDao { private final JdbcTemplate jdbcTemplate; private static final ResultSetExtractor<List<OrderMessage>> M = r -> { List<OrderMessage> 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 List<OrderMessage> selectPendingRecords(LocalDateTime start, LocalDateTime end, List<Integer> 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@RequiredArgsConstructorpublic class OrderMessageService { private static final Logger LOGGER = LoggerFactory.getLogger(OrderMessageService.class); private final OrderMessageDao orderMessageDao; private static final List<Integer> 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); List<OrderMessage> 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@DisallowConcurrentExecutionpublic 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=false&characterEncoding=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); }}這個(gè)例子里面用了 create_time 做輪詢，實(shí)際上可以添加一個(gè)調(diào)度時(shí)間 schedule_time 列做輪詢，這樣子才能更容易定制空閑時(shí)和忙碌時(shí)候的調(diào)度策略。上面的示例的運(yùn)行效果如下：
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.111: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 triggers11:58:27.217 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.simpl.PropertySettingJobFactory - Producing instance of Job 'delayJob.orderMessageDelayJob', class=club.throwable.jdbc.OrderMessageDelayJob11:58:27.219 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@10eb8c5311:58:27.220 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 0 triggers11:58:27.221 [DefaultQuartzScheduler_Worker-1] DEBUG org.quartz.core.JobRunShell - Calling execute on job delayJob.orderMessageDelayJob11: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@3d27ece411:58:34.459 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@64e808af11:58:34.470 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@79c8c2b711:58:34.477 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@19a6236911:58:34.485 [HikariPool-1 connection adder] DEBUG com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection com.mysql.jdbc.JDBC4Connection@1673d01711: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 query11: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 DataSource11: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更新為111:58:35.342 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL update11: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 DataSource11:58:35.354 [DefaultQuartzScheduler_Worker-1] INFO club.throwable.jdbc.OrderMessageService - 處理訂單[10087],狀態(tài)由0更新為111:58:35.355 [DefaultQuartzScheduler_Worker-1] DEBUG org.springframework.jdbc.core.JdbcTemplate - Executing prepared SQL update11: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 DataSource11: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 triggers11:58:37.206 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.simpl.PropertySettingJobFactory - Producing instance of Job 'delayJob.orderMessageDelayJob', class=club.throwable.jdbc.OrderMessageDelayJob11:58:37.206 [DefaultQuartzScheduler_QuartzSchedulerThread] DEBUG org.quartz.core.QuartzSchedulerThread - batch acquisition of 0 triggers
RabbitMQ死信隊(duì)列
使用 RabbitMQ 死信隊(duì)列依賴于 RabbitMQ 的兩個(gè)特性： TTL 和 DLX 。
TTL ： Time To Live ，消息存活時(shí)間，包括兩個(gè)維度：隊(duì)列消息存活時(shí)間和消息本身的存活時(shí)間。
DLX ： Dead Letter Exchange ，死信交換器。
畫個(gè)圖描述一下這兩個(gè)特性：
下面為了簡單起見， TTL 使用了針對(duì)隊(duì)列的維度。引入 RabbitMQ 的Java驅(qū)動(dòng)：
<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"); Map<String, 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 timestamp; 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 timestamp; } public String getDescription() { return description; } }}運(yùn)行 main() 方法結(jié)果如下：
16:35:58.638 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:1008616:35:58.641 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:1008716:35:58.641 [main] INFO club.throwable.dlx.DlxMain - 發(fā)送消息成功,訂單ID:1008816:36:03.646 [pool-1-thread-4] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10086],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:5816:36:03.670 [pool-1-thread-5] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10087],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:5816:36:03.670 [pool-1-thread-6] INFO club.throwable.dlx.DlxMain - 處理消息成功:訂單[10088],訂單創(chuàng)建時(shí)間為:2019-08-20 16:35:58
時(shí)間輪
時(shí)間輪 TimingWheel 是一種高效、低延遲的調(diào)度數(shù)據(jù)結(jié)構(gòu)，底層采用數(shù)組實(shí)現(xiàn)存儲(chǔ)任務(wù)列表的環(huán)形隊(duì)列，示意圖如下：
這里暫時(shí)不對(duì)時(shí)間輪和其實(shí)現(xiàn)作分析，只簡單舉例說明怎么使用時(shí)間輪實(shí)現(xiàn)延時(shí)任務(wù)。這里使用 Netty 提供的 HashedWheelTimer ，引入依賴：
<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 timestamp; 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)); } }}運(yùn)行結(jié)果：
任務(wù)執(zhí)行時(shí)間:2019-08-20 17:19:49.310,訂單創(chuàng)建時(shí)間:2019-08-20 17:19:43.294,訂單ID:10086
任務(wù)執(zhí)行時(shí)間:2019-08-20 17:19:54.297,訂單創(chuàng)建時(shí)間:2019-08-20 17:19:43.301,訂單ID:10087
任務(wù)執(zhí)行時(shí)間:2019-08-20 17:19:59.297,訂單創(chuàng)建時(shí)間:2019-08-20 17:19:43.301,訂單ID:10088
一般來說，任務(wù)執(zhí)行的時(shí)候應(yīng)該使用另外的業(yè)務(wù)線程池，以免阻塞時(shí)間輪本身的運(yùn)動(dòng)。
選用的方案實(shí)現(xiàn)過程
最終選用了基于 Redis 的有序集合 Sorted Set 和 Quartz 短輪詢進(jìn)行實(shí)現(xiàn)。具體方案是：
Sorted Set 的member和score添加到訂單隊(duì)列 Sorted Set 中。JSON 字符串分別作為field和value添加到訂單隊(duì)列內(nèi)容 Hash 中。Lua 腳本保證原子性。Sorted Set 的命令 ZREVRANGEBYSCORE 彈出指定數(shù)量的訂單ID對(duì)應(yīng)的訂單隊(duì)列內(nèi)容 Hash 中的訂單推送內(nèi)容數(shù)據(jù)進(jìn)行處理。對(duì)于第4點(diǎn)處理有兩種方案：
ZREVRANGEBYSCORE 、 ZREM 和 HDEL 命令要在同一個(gè) Lua 腳本中執(zhí)行，這樣的話 Lua 腳本的編寫難度大，并且由于彈出數(shù)據(jù)已經(jīng)在 Redis 中刪除，如果數(shù)據(jù)處理失敗則可能需要從數(shù)據(jù)庫重新查詢補(bǔ)償。Sorted Set 和訂單隊(duì)列內(nèi)容 Hash 中對(duì)應(yīng)的數(shù)據(jù)，這樣的話需要控制并發(fā)，有重復(fù)執(zhí)行的可能性。最終暫時(shí)選用了方案一，也就是從 Sorted Set 彈出訂單ID并且從 Hash 中獲取完推送數(shù)據(jù)之后馬上刪除這兩個(gè)集合中對(duì)應(yīng)的數(shù)據(jù)。方案的流程圖大概是這樣：
這里先詳細(xì)說明一下用到的 Redis 命令。
Sorted Set相關(guān)命令
ZADD 命令 - 將一個(gè)或多個(gè)成員元素及其分?jǐn)?shù)值加入到有序集當(dāng)中。
ZADD KEY SCORE1 VALUE1.. SCOREN VALUEN
ZREVRANGEBYSCORE 命令 - 返回有序集中指定分?jǐn)?shù)區(qū)間內(nèi)的所有的成員。有序集成員按分?jǐn)?shù)值遞減(從大到小)的次序排列。
ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]
max：分?jǐn)?shù)區(qū)間 - 最大分?jǐn)?shù)。 min：分?jǐn)?shù)區(qū)間 - 最小分?jǐn)?shù)。 WITHSCORES：可選參數(shù)，是否返回分?jǐn)?shù)值，指定則會(huì)返回得分值。 LIMIT：可選參數(shù)，offset和count原理和 MySQL 的 LIMIT offset,size 一致，如果不指定此參數(shù)則返回整個(gè)集合的數(shù)據(jù)。 ZREM 命令 - 用于移除有序集中的一個(gè)或多個(gè)成員，不存在的成員將被忽略。
ZREM key member [member ...]
Hash相關(guān)命令 HMSET 命令 - 同時(shí)將多個(gè)field-value(字段-值)對(duì)設(shè)置到哈希表中。
HMSET KEY_NAME FIELD1 VALUE1 ...FIELDN VALUEN
HDEL 命令 - 刪除哈希表key中的一個(gè)或多個(gè)指定字段，不存在的字段將被忽略。
HDEL KEY_NAME FIELD1.. FIELDN
Lua相關(guān) 加載 Lua 腳本并且返回腳本的 SHA-1 字符串： SCRIPT LOAD script 。 執(zhí)行已經(jīng)加載的 Lua 腳本： EVALSHA sha1 numkeys key [key ...] arg [arg ...] 。 unpack 函數(shù)可以把 table 類型的參數(shù)轉(zhuǎn)化為可變參數(shù)，不過需要注意的是 unpack 函數(shù)必須使用在非變量定義的函數(shù)調(diào)用的最后一個(gè)參數(shù)，否則會(huì)失效，詳細(xì)見 Stackoverflow 的提問 table.unpack() only returns the first element 。
PS：如果不熟悉Lua語言，建議系統(tǒng)學(xué)習(xí)一下，因?yàn)橄胗煤肦edis，一定離不開Lua。
引入依賴：
<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 腳本 /lua/enqueue.lua 和 /lua/dequeue.lua ：
-- /lua/enqueue.lualocal 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 endendreturn nil編寫核心API代碼：
// Jedis提供者@Componentpublic class JedisProvider implements InitializingBean { private JedisPool jedisPool; @Override public void afterPropertiesSet() throws Exception { jedisPool = new JedisPool(); } public Jedis provide(){ return jedisPool.getResource(); }}// OrderMessage@Datapublic class OrderMessage { private String orderId; private BigDecimal amount; private Long userId;}// 延遲隊(duì)列接口public interface OrderDelayQueue { void enqueue(OrderMessage message); List<OrderMessage> dequeue(String min, String max, String offset, String limit); List<OrderMessage> dequeue(); String enqueueSha(); String dequeueSha();}// 延遲隊(duì)列實(shí)現(xiàn)類@RequiredArgsConstructor@Componentpublic 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 AtomicReference<String> ENQUEUE_LUA_SHA = new AtomicReference<>(); private static final AtomicReference<String> DEQUEUE_LUA_SHA = new AtomicReference<>(); private static final List<String> KEYS = Lists.newArrayList(); private final JedisProvider jedisProvider; static { KEYS.add(ORDER_QUEUE); KEYS.add(ORDER_DETAIL_QUEUE); } @Override public void enqueue(OrderMessage message) { List<String> 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 List<OrderMessage> dequeue() { // 30分鐘之前 String maxScore = String.valueOf(System.currentTimeMillis() - 30 * 60 * 1000); return dequeue(MIN_SCORE, maxScore, OFFSET, LIMIT); } @SuppressWarnings("unchecked") @Override public List<OrderMessage> dequeue(String min, String max, String offset, String limit) { List<String> args = new ArrayList<>(); args.add(max); args.add(min); args.add(offset); args.add(limit); List<OrderMessage> result = Lists.newArrayList(); try (Jedis jedis = jedisProvider.provide()) { List<String> eval = (List<String>) 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(); // 寫入測試數(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)); List<String> args = Lists.newArrayList(); args.add(message.getOrderId()); // 測試需要,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); } List<OrderMessage> dequeue = queue.dequeue(); System.out.println(dequeue); }}這里先執(zhí)行一次 main() 方法驗(yàn)證一下延遲隊(duì)列是否生效：
[OrderMessage(orderId=ORDER_ID_10086, amount=10086, userId=10086, timestamp=2019-08-21 08:32:22.885)]
確定延遲隊(duì)列的代碼沒有問題，接著編寫一個(gè) Quartz 的 Job 類型的消費(fèi)者 OrderMessageConsumer ：
@DisallowConcurrentExecution@Componentpublic 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í)行......"); List<OrderMessage> messages = orderDelayQueue.dequeue(); if (!messages.isEmpty()) { // 簡單的列表等分放到線程池中執(zhí)行 List<List<OrderMessage>> partition = Lists.partition(messages, 2); int size = partition.size(); final CountDownLatch latch = new CountDownLatch(size); for (List<OrderMessage> 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 List<OrderMessage> 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(); } } }}上面的消費(fèi)者設(shè)計(jì)的時(shí)候需要有以下考量：
@DisallowConcurrentExecution 注解不允許 Job 并發(fā)執(zhí)行，其實(shí)多個(gè) Job 并發(fā)執(zhí)行意義不大，因?yàn)槲覀儾捎玫氖嵌涕g隔的輪詢，而 Redis 是單線程處理命令，在客戶端做多線程其實(shí)效果不佳。BUSINESS_WORKER_POOL 的線程容量或者隊(duì)列應(yīng)該綜合 LIMIT 值、等分訂單信息列表中使用的 size 值以及 ConsumeTask 里面具體的執(zhí)行時(shí)間進(jìn)行考慮，這里只是為了方便使用了固定容量的線程池。ConsumeTask 中應(yīng)該對(duì)每一條訂單信息的處理單獨(dú)捕獲異常和吞并異常，或者把處理單個(gè)訂單信息的邏輯封裝成一個(gè)不拋出異常的方法。其他 Quartz 相關(guān)的代碼：
// Quartz配置類@Configurationpublic 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; } }}這里暫時(shí)使用了內(nèi)存態(tài)的 RAMJobStore 去存放任務(wù)和觸發(fā)器的相關(guān)信息，如果在生產(chǎn)環(huán)境最好替換成基于 MySQL 也就是 JobStoreTX 進(jìn)行集群化，最后是啟動(dòng)函數(shù)和 CommandLineRunner 的實(shí)現(xiàn)：
@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)備一些測試數(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()) { List<OrderMessage> 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 Map<String, Double> map = Maps.newHashMap(); Map<String, 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); } }}輸出結(jié)果如下：
2019-08-21 22:45:59.518 INFO 33000 --- [ryBean_Worker-1] club.throwable.OrderMessageConsumer : 訂單消息處理定時(shí)任務(wù)開始執(zhí)行......
2019-08-21 22:45:59.525 INFO 33000 --- [onsumerWorker-4] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_91, amount=91, userId=91, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.525 INFO 33000 --- [onsumerWorker-2] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_95, amount=95, userId=95, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.525 INFO 33000 --- [onsumerWorker-1] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_97, amount=97, userId=97, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.525 INFO 33000 --- [onsumerWorker-0] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_99, amount=99, userId=99, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.525 INFO 33000 --- [onsumerWorker-3] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_93, amount=93, userId=93, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.539 INFO 33000 --- [onsumerWorker-2] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_94, amount=94, userId=94, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.539 INFO 33000 --- [onsumerWorker-1] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_96, amount=96, userId=96, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.539 INFO 33000 --- [onsumerWorker-3] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_92, amount=92, userId=92, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.539 INFO 33000 --- [onsumerWorker-0] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_98, amount=98, userId=98, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.539 INFO 33000 --- [onsumerWorker-4] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_90, amount=90, userId=90, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:45:59.540 INFO 33000 --- [ryBean_Worker-1] club.throwable.OrderMessageConsumer : 訂單消息處理定時(shí)任務(wù)執(zhí)行完畢,耗時(shí):22 ms......
2019-08-21 22:46:04.515 INFO 33000 --- [ryBean_Worker-2] club.throwable.OrderMessageConsumer : 訂單消息處理定時(shí)任務(wù)開始執(zhí)行......
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-5] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_89, amount=89, userId=89, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-6] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_87, amount=87, userId=87, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-7] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_85, amount=85, userId=85, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-5] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_88, amount=88, userId=88, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-2] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_83, amount=83, userId=83, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-1] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_81, amount=81, userId=81, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-6] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_86, amount=86, userId=86, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-2] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_82, amount=82, userId=82, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-7] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_84, amount=84, userId=84, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [onsumerWorker-1] club.throwable.OrderMessageConsumer : 處理訂單信息,內(nèi)容:OrderMessage(orderId=ORDER_ID_80, amount=80, userId=80, timestamp=2019-08-21 22:45:59.475)
2019-08-21 22:46:04.516 INFO 33000 --- [ryBean_Worker-2] club.throwable.OrderMessageConsumer : 訂單消息處理定時(shí)任務(wù)執(zhí)行完畢,耗時(shí):1 ms......
......
首次執(zhí)行的時(shí)候涉及到一些組件的初始化，會(huì)比較慢，后面看到由于我們只是簡單打印訂單信息，所以定時(shí)任務(wù)執(zhí)行比較快。如果在不調(diào)整當(dāng)前架構(gòu)的情況下，生產(chǎn)中需要注意：
JobStore 為 JDBC 模式， Quartz 官方有完整教程，或者看筆者之前翻譯的 Quartz 文檔。這里其實(shí)有一個(gè)性能隱患，命令 ZREVRANGEBYSCORE 的時(shí)間復(fù)雜度可以視為為 O(N) ， N 是集合的元素個(gè)數(shù)，由于這里把所有的訂單信息都放進(jìn)了同一個(gè) Sorted Set ( ORDER_QUEUE )中，所以在一直有新增數(shù)據(jù)的時(shí)候， dequeue 腳本的時(shí)間復(fù)雜度一直比較高，后續(xù)訂單量升高之后會(huì)此處一定會(huì)成為性能瓶頸，后面會(huì)給出解決的方案。
小結(jié)
這篇文章主要從一個(gè)實(shí)際生產(chǎn)案例的仿真例子入手，分析了當(dāng)前延時(shí)任務(wù)的一些實(shí)現(xiàn)方案，還基于 Redis 和 Quartz 給出了一個(gè)完整的示例。當(dāng)前的示例只是處于可運(yùn)行的狀態(tài)，有些問題尚未解決。下一篇文章會(huì)著眼于解決兩個(gè)方面的問題：
還有一點(diǎn)， 架構(gòu)是基于業(yè)務(wù)形態(tài)演進(jìn)出來的，很多東西需要結(jié)合場景進(jìn)行方案設(shè)計(jì)和改進(jìn)，思路僅供參考，切勿照搬代碼 。
以上所述是小編給大家介紹的使用Redis實(shí)現(xiàn)延時(shí)任務(wù)的解決方案，非常不錯(cuò)，具有一定的參考借鑒價(jià)值，需要的朋友參考下吧！
新聞熱點(diǎn)
疑難解答
圖片精選
