Redis延遲隊列和分佈式延遲隊列的簡答實現
最近,又重新學習瞭下Redis,Redis不僅能快還能慢,簡直利器,今天就為大傢介紹一下Redis延遲隊列和分佈式延遲隊列的簡單實現。
在我們的工作中,很多地方使用延遲隊列,比如訂單到期沒有付款取消訂單,制訂一個提醒的任務等都需要延遲隊列,那麼我們需要實現延遲隊列。我們本文的梗概如下,同學們可以選擇性閱讀。
1. 實現一個簡單的延遲隊列。
我們知道目前JAVA可以有DelayedQueue,我們首先開一個DelayQueue的結構類圖。DelayQueue實現瞭Delay、BlockingQueue接口。也就是DelayQueue是一種阻塞隊列。
我們在看一下Delay的類圖。Delayed接口也實現瞭Comparable接口,也就是我們使用Delayed的時候需要實現CompareTo方法。因為隊列中的數據需要排一下先後,根據我們自己的實現。Delayed接口裡邊有一個方法就是getDelay方法,用於獲取延遲時間,判斷是否時間已經到瞭延遲的時間,如果到瞭延遲的時間就可以從隊列裡邊獲取瞭。
我們創建一個Message類,實現瞭Delayed接口,我們主要把getDelay和compareTo進行實現。在Message的構造方法的地方傳入延遲的時間,單位是毫秒,計算好觸發時間fireTime。同時按照延遲時間的升序進行排序。我重寫瞭裡邊的toString方法,用於將Message按照我寫的方法進行輸出。
package com.hqs.delayQueue.bean; import java.util.concurrent.BlockingQueue; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; /** * @author huangqingshi * @Date 2020-04-18 */ public class Message implements Delayed { private String body; private long fireTime; public String getBody() { return body; } public long getFireTime() { return fireTime; } public Message(String body, long delayTime) { this.body = body; this.fireTime = delayTime + System.currentTimeMillis(); } public long getDelay(TimeUnit unit) { return unit.convert(this.fireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS); } public int compareTo(Delayed o) { return (int) (this.getDelay(TimeUnit.MILLISECONDS) -o.getDelay(TimeUnit.MILLISECONDS)); } @Override public String toString() { return System.currentTimeMillis() + ":" + body; } public static void main(String[] args) throws InterruptedException { System.out.println(System.currentTimeMillis() + ":start"); BlockingQueue<Message> queue = new DelayQueue<>(); Message message1 = new Message("hello", 1000 * 5L); Message message2 = new Message("world", 1000 * 7L); queue.put(message1); queue.put(message2); while (queue.size() > 0) { System.out.println(queue.take()); } } }
裡邊的main方法裡邊聲明瞭兩個Message,一個延遲5秒,一個延遲7秒,時間到瞭之後會將接取出並且打印。輸出的結果如下,正是我們所期望的。
1587218430786:start
1587218435789:hello
1587218437793:world
這個方法實現起來真的非常簡單。但是缺點也是很明顯的,就是數據在內存裡邊,數據比較容易丟失。那麼我們需要采用Redis實現分佈式的任務處理。
2. 使用Redis的list實現分佈式延遲隊列。
本地需要安裝一個Redis,我自己是使用Docker構建一個Redis,非常快速,命令也沒多少。我們直接啟動Redis並且暴露6379端口。進入之後直接使用客戶端命令即可查看和調試數據。
docker pull redis docker run -itd --name redisLocal -p 6379:6379 redis docker exec -it redisLocal /bin/bash redis-cli
我本地采用spring-boot的方式連接redis,pom文件列一下,供大傢參考。
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.6.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.hqs</groupId> <artifactId>delayQueue</artifactId> <version>0.0.1-SNAPSHOT</version> <name>delayQueue</name> <description>Demo project for Spring Boot</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
加上Redis的配置放到application.properties裡邊即可實現Redis連接,非常的方便。
# redis redis.host=127.0.0.1 redis.port=6379 redis.password= redis.maxIdle=100 redis.maxTotal=300 redis.maxWait=10000 redis.testOnBorrow=true redis.timeout=100000
接下來實現一個基於Redis的list數據類型進行實現的一個類。我們使用RedisTemplate操作Redis,這個裡邊封裝好我們所需要的Redis的一些方法,用起來非常方便。這個類允許延遲任務做多有10W個,也是避免數據量過大對Redis造成影響。如果在線上使用的時候也需要考慮延遲任務的多少。太多幾百萬幾千萬的時候可能數據量非常大,我們需要計算Redis的空間是否夠。這個代碼也是非常的簡單,一個用於存放需要延遲的消息,采用offer的方法。另外一個是啟動一個線程, 如果消息時間到瞭,那麼就將數據lpush到Redis裡邊。
package com.hqs.delayQueue.cache; import com.hqs.delayQueue.bean.Message; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.RedisTemplate; import java.util.concurrent.BlockingQueue; /** * @author huangqingshi * @Date 2020-04-18 */ @Slf4j public class RedisListDelayedQueue{ private static final int MAX_SIZE_OF_QUEUE = 100000; private RedisTemplate<String, String> redisTemplate; private String queueName; private BlockingQueue<Message> delayedQueue; public RedisListDelayedQueue(RedisTemplate<String, String> redisTemplate, String queueName, BlockingQueue<Message> delayedQueue) { this.redisTemplate = redisTemplate; this.queueName = queueName; this.delayedQueue = delayedQueue; init(); } public void offerMessage(Message message) { if(delayedQueue.size() > MAX_SIZE_OF_QUEUE) { throw new IllegalStateException("超過隊列要求最大值,請檢查"); } try { log.info("offerMessage:" + message); delayedQueue.offer(message); } catch (Exception e) { log.error("offMessage異常", e); } } public void init() { new Thread(() -> { while(true) { try { Message message = delayedQueue.take(); redisTemplate.opsForList().leftPush(queueName, message.toString()); } catch (InterruptedException e) { log.error("取消息錯誤", e); } } }).start(); } }
接下來我們看一下,我們寫一個測試的controller。大傢看一下這個請求/redis/listDelayedQueue的代碼位置。我們也是生成瞭兩個消息,然後把消息放到隊列裡邊,另外我們在啟動一個線程任務,用於將數據從Redis的list中獲取。方法也非常簡單。
package com.hqs.delayQueue.controller; import com.hqs.delayQueue.bean.Message; import com.hqs.delayQueue.cache.RedisListDelayedQueue; import com.hqs.delayQueue.cache.RedisZSetDelayedQueue; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.ResponseBody; import java.util.Set; import java.util.concurrent.*; /** * @author huangqingshi * @Date 2020-04-18 */ @Slf4j @Controller public class DelayQueueController { private static final int CORE_SIZE = Runtime.getRuntime().availableProcessors(); //註意RedisTemplate用的String,String,後續所有用到的key和value都是String的 @Autowired RedisTemplate<String, String> redisTemplate; private static ThreadPoolExecutor taskExecPool = new ThreadPoolExecutor(CORE_SIZE, CORE_SIZE, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>()); @GetMapping("/redisTest") @ResponseBody public String redisTest() { redisTemplate.opsForValue().set("a","b",60L, TimeUnit.SECONDS); System.out.println(redisTemplate.opsForValue().get("a")); return "s"; } @GetMapping("/redis/listDelayedQueue") @ResponseBody public String listDelayedQueue() { Message message1 = new Message("hello", 1000 * 5L); Message message2 = new Message("world", 1000 * 7L); String queueName = "list_queue"; BlockingQueue<Message> delayedQueue = new DelayQueue<>(); RedisListDelayedQueue redisListDelayedQueue = new RedisListDelayedQueue(redisTemplate, queueName, delayedQueue); redisListDelayedQueue.offerMessage(message1); redisListDelayedQueue.offerMessage(message2); asyncListTask(queueName); return "success"; } @GetMapping("/redis/zSetDelayedQueue") @ResponseBody public String zSetDelayedQueue() { Message message1 = new Message("hello", 1000 * 5L); Message message2 = new Message("world", 1000 * 7L); String queueName = "zset_queue"; BlockingQueue<Message> delayedQueue = new DelayQueue<>(); RedisZSetDelayedQueue redisZSetDelayedQueue = new RedisZSetDelayedQueue(redisTemplate, queueName, delayedQueue); redisZSetDelayedQueue.offerMessage(message1); redisZSetDelayedQueue.offerMessage(message2); asyncZSetTask(queueName); return "success"; } public void asyncListTask(String queueName) { taskExecPool.execute(() -> { for(;;) { String message = redisTemplate.opsForList().rightPop(queueName); if(message != null) { log.info(message); } } }); } public void asyncZSetTask(String queueName) { taskExecPool.execute(() -> { for(;;) { Long nowTimeInMs = System.currentTimeMillis(); System.out.println("nowTimeInMs:" + nowTimeInMs); Set<String> messages = redisTemplate.opsForZSet().rangeByScore(queueName, 0, nowTimeInMs); if(messages != null && messages.size() != 0) { redisTemplate.opsForZSet().removeRangeByScore(queueName, 0, nowTimeInMs); for (String message : messages) { log.info("asyncZSetTask:" + message + " " + nowTimeInMs); } log.info(redisTemplate.opsForZSet().zCard(queueName).toString()); } try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } } }); } }
我就不把運行結果寫出來瞭,感興趣的同學自己自行試驗。當然這個方法也是從內存中拿出數據,到時間之後放到Redis裡邊,還是會存在程序啟動的時候,任務進行丟失。我們繼續看另外一種方法更好的進行這個問題的處理。
3.使用Redis的zSet實現分佈式延遲隊列。
我們需要再寫一個ZSet的隊列處理。下邊的offerMessage主要是把消息直接放入緩存中。采用Redis的ZSET的zadd方法。zadd(key, value, score) 即將key=value的數據賦予一個score, 放入緩存中。score就是計算出來延遲的毫秒數。
package com.hqs.delayQueue.cache; import com.hqs.delayQueue.bean.Message; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.RedisTemplate; import java.util.concurrent.BlockingQueue; /** * @author huangqingshi * @Date 2020-04-18 */ @Slf4j public class RedisZSetDelayedQueue { private static final int MAX_SIZE_OF_QUEUE = 100000; private RedisTemplate<String, String> redisTemplate; private String queueName; private BlockingQueue<Message> delayedQueue; public RedisZSetDelayedQueue(RedisTemplate<String, String> redisTemplate, String queueName, BlockingQueue<Message> delayedQueue) { this.redisTemplate = redisTemplate; this.queueName = queueName; this.delayedQueue = delayedQueue; } public void offerMessage(Message message) { if(delayedQueue.size() > MAX_SIZE_OF_QUEUE) { throw new IllegalStateException("超過隊列要求最大值,請檢查"); } long delayTime = message.getFireTime() - System.currentTimeMillis(); log.info("zset offerMessage" + message + delayTime); redisTemplate.opsForZSet().add(queueName, message.toString(), message.getFireTime()); } }
上邊的Controller方法已經寫好瞭測試的方法。/redis/zSetDelayedQueue,裡邊主要使用ZSet的zRangeByScore(key, min, max)。主要是從score從0,當前時間的毫秒數獲取。取出數據後再采用removeRangeByScore,將數據刪除。這樣數據可以直接寫到Redis裡邊,然後取出數據後直接處理。這種方法比前邊的方法稍微好一些,但是實際上還存在一些問題,因為依賴Redis,如果Redis內存不足或者連不上的時候,系統將變得不可用。
4. 總結一下,另外還有哪些可以延遲隊列。
上面的方法其實還是存在問題的,比如系統重啟的時候還是會造成任務的丟失。所以我們在生產上使用的時候,我們還需要將任務保存起來,比如放到數據庫和文件存儲系統將數據存儲起來,這樣做到double-check,雙重檢查,最終達到任務的99.999%能夠處理。
其實還有很多東西可以實現延遲隊列。
1) RabbitMQ就可以實現此功能。這個消息隊列可以把數據保存起來並且進行處理。
2)Kafka也可以實現這個功能。
3)Netty的HashedWheelTimer也可以實現這個功能。
最後放上我的代碼: https://github.com/stonehqs/delayQueue
到此這篇關於Redis延遲隊列和分佈式延遲隊列的簡答實現的文章就介紹到這瞭,更多相關Redis延遲隊列和分佈式延遲隊列內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- springboot 集成redis哨兵主從的實現
- Redis優雅地實現延遲隊列的方法分享
- Springboot/Springcloud項目集成redis進行存取的過程解析
- springboot整合使用雲服務器上的Redis方法
- Spring中RedisTemplate的基本使用淺析