日常需求開發過程中，不免會遇到需要通過代碼進行異步處理的情況，比如批量發送郵件，批量發送短信，數據導入，為了減少用戶的等待，不希望一直菊花轉啊轉，因此需要進行異步處理，做法就是講要處理的數據添加到隊列當中，然后按照排隊的先后順序進行異步處理。
這個隊列，可以是專業的消息隊列，如 RocketMQ/RabbitMQ 等，一般項目中，如果只是為了進行異步，未免有點殺雞用牛刀的意味。
也可以使用基于 JVM 內存實現隊列，但是如果項目進行了重啟，就會造成隊列數據丟失。
大部分的項目都會用到 Redis 中間件作為緩存使用，此時使用 Redis 的 list 結構來實現隊列則是非常合適的選擇。
因此，本文主要講解基于 Redis 的方式實現異步隊列。
本文首發個人技術博客： https://nullpointer.pw/redis-block-queue.html
基于 Redis 的 list 實現隊列的方式也有多種，先說第一種不推薦的方式，即使用LPUSH生產消息，然后 while(true) 中通過RPOP消費消息，這種方式的確可以實現，但是不斷代碼不斷的輪詢，勢必會消耗一些系統的資源。
第二種方式也是不推薦的方式，也是通過 LPUSH生產消息，然后通過 BRPOP 進行阻塞地等待并消費消息，這種方式較第一種方式減少了無用的輪詢，降低系統資源的消耗，但是可能會存在隊列消息丟失的情況，如果取出了消息然后處理失敗，這個被取出的消息就將丟失。
第二種方式就是下文要介紹的方式，首先也是通過 LPUSH 生產消息，然后通過 BRPOPLPUSH阻塞地等待 list 新消息到來，有了新消息才開始消費，同時將消息備份到另外一個 list 當中，這種方式具備了第二種方式的優點，即減少了無用的輪詢，同時也對消息進行了備份不會丟失數據，如果處理成功，可以通過 LREM 對備份的 list 中當前的這條消息進行刪除處理。這種方式實現方式可以參考 模式： 安全的隊列 .
Redis 基礎
|
1
2
3
4
5
6
7
8
|
# 將一個或多個值 value 插入到列表 key 的表頭LPUSH key value [value …]# 阻塞式等待，將列表 source 中的最后一個元素 (尾元素) 彈出，并返回給客戶端。將 source 彈出的元素插入到列表 destination ，作為 destination 列表的的頭元素。超時參數 timeout 接受一個以秒為單位的數字作為值。超時參數設為 0 表示阻塞時間可以無限期延長 (block indefinitely) 。BRPOPLPUSH source destination timeout# 根據參數 count 的值，移除列表中與參數 value 相等的元素。LREM key count value |
代碼實現隊列消息生產者
筆者使用的是 Spring 相關 API 實現對 Redis 指令的調用。首先實現消息的生產代碼，封裝到一個工具類方法當中。這里很簡單，就是調用了 lpush 方法，將序列化的 key 和 value 添加到列表當中去。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
@Resourceprivate RedisConnectionFactory connectionFactory;public void lPush(@Nonnull String key, @Nonnull String value) { RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory); try { byte[] byteKey = RedisSerializer.string().serialize(getKey(key)); byte[] byteValue = RedisSerializer.string().serialize(value); assert byteKey != null; connection.lPush(byteKey, byteValue); } finally { RedisConnectionUtils.releaseConnection(connection, connectionFactory); }} |
代碼實現隊列消息消費者
因為實現隊列消費消息的代碼比較多，不可能每個需要阻塞消費的地方，對需要寫這一坨代碼，因此使用 Java8 的函數式接口實現方法的傳遞，同時阻塞式獲取消息代碼使用新線程去執行。
有人看到以下代碼要吐槽了，不是說不用 while(true) 嗎，怎么你這里面還是有，這里稍微解釋一下，因為 SpringBoot 一般會指定 timeout 的全局超時時間，即使 BRPOPLPUSH 設置了 0，即無限期，當超出了 timeout 設置的值時，就會拋出 QueryTimeoutException 異常導致線程退出，因此添加了 try/catch 對異常進行捕獲并忽略，同時使用 while(true) 保證線程可以繼續執行。
代碼中記錄了當前消息處理結果，如果處理結果為成功，需要對備份隊列的當前消息進行刪除。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public void bRPopLPush(@Nonnull String key, Consumer<String> consumer) { CompletableFuture.runAsync(() -> { RedisConnection connection = RedisConnectionUtils.getConnection(connectionFactory); try { byte[] srcKey = RedisSerializer.string().serialize(getKey(key)); byte[] dstKey = RedisSerializer.string().serialize(getBackupKey(key)); assert srcKey != null; assert dstKey != null; while (true) { byte[] byteValue = new byte[0]; boolean success = false; try { byteValue = connection.bRPopLPush(0, srcKey, dstKey); if (byteValue != null && byteValue.length != 0) { consumer.accept(new String(byteValue)); success = true; } } catch (Exception ignored) { // 防止獲取 key 達到超時時間拋出 QueryTimeoutException 異常退出 } finally { if (success) { // 處理成功才刪除備份隊列的 key connection.lRem(dstKey, 1, byteValue); } } } } finally { RedisConnectionUtils.releaseConnection(connection, connectionFactory); } });} |
測試代碼
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
@Testpublic void testLPush() throws InterruptedException { String queueA = "queueA"; int i = 0; while (true) { String msg = "Hello-" + i++; redisBlockQueue.lPush(queueA, msg); System.out.println("lPush: " + msg); Thread.sleep(3000); }}@Testpublic void testBRPopLPush() { String queueA = "queueA"; redisBlockQueue.bRPopLPush(queueA, (val) -> { // 在這里處理具體的業務邏輯 System.out.println("val: " + val); }); // 防止 Junit 進程退出 LockSupport.park();} |
項目使用方式
為了方便使用，我將其抽取為了一個工具類，使用時通過 Spring 注入使用即可，
隊列消費可以使用如下方式在項目啟動的時候就進行阻塞監聽隊列，等待消費
|
1
2
3
4
5
6
7
8
9
|
@Resourceprivate RedisBlockQueue redisBlockQueue;@PostConstructpublic void init() { redisBlockQueue.bRPopLPush(xx, (value) -> { //... });} |
本文完整代碼下載github 地址
到此這篇關于基于Redis實現阻塞隊列的文章就介紹到這了,更多相關Redis阻塞隊列內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://www.cnblogs.com/vcmq/p/13509269.html
