主頁 > 知識庫 > Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)

Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)

熱門標簽:魔獸2青云地圖標注 日本中國地圖標注 貴州電銷卡外呼系統(tǒng) 鄭州人工智能電銷機器人系統(tǒng) 十堰營銷電銷機器人哪家便宜 山東外呼銷售系統(tǒng)招商 超呼電話機器人 北京400電話辦理收費標準 宿遷便宜外呼系統(tǒng)平臺

        最近,又重新學習了下Redis,Redis不僅能快還能慢,簡直利器,今天就為大家介紹一下Redis延遲隊列和分布式延遲隊列的簡單實現(xiàn)。

  在我們的工作中,很多地方使用延遲隊列,比如訂單到期沒有付款取消訂單,制訂一個提醒的任務等都需要延遲隊列,那么我們需要實現(xiàn)延遲隊列。我們本文的梗概如下,同學們可以選擇性閱讀。

1. 實現(xiàn)一個簡單的延遲隊列。

  我們知道目前JAVA可以有DelayedQueue,我們首先開一個DelayQueue的結(jié)構(gòu)類圖。DelayQueue實現(xiàn)了Delay、BlockingQueue接口。也就是DelayQueue是一種阻塞隊列。

  我們在看一下Delay的類圖。Delayed接口也實現(xiàn)了Comparable接口,也就是我們使用Delayed的時候需要實現(xiàn)CompareTo方法。因為隊列中的數(shù)據(jù)需要排一下先后,根據(jù)我們自己的實現(xiàn)。Delayed接口里邊有一個方法就是getDelay方法,用于獲取延遲時間,判斷是否時間已經(jīng)到了延遲的時間,如果到了延遲的時間就可以從隊列里邊獲取了。

  我們創(chuàng)建一個Message類,實現(xiàn)了Delayed接口,我們主要把getDelay和compareTo進行實現(xiàn)。在Message的構(gòu)造方法的地方傳入延遲的時間,單位是毫秒,計算好觸發(fā)時間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");
        BlockingQueueMessage> 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秒,時間到了之后會將接取出并且打印。輸出的結(jié)果如下,正是我們所期望的。

1587218430786:start
1587218435789:hello
1587218437793:world

  這個方法實現(xiàn)起來真的非常簡單。但是缺點也是很明顯的,就是數(shù)據(jù)在內(nèi)存里邊,數(shù)據(jù)比較容易丟失。那么我們需要采用Redis實現(xiàn)分布式的任務處理。

  2. 使用Redis的list實現(xiàn)分布式延遲隊列。

  本地需要安裝一個Redis,我自己是使用Docker構(gòu)建一個Redis,非??焖伲钜矝]多少。我們直接啟動Redis并且暴露6379端口。進入之后直接使用客戶端命令即可查看和調(diào)試數(shù)據(jù)。

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里邊即可實現(xiàn)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

  接下來實現(xiàn)一個基于Redis的list數(shù)據(jù)類型進行實現(xiàn)的一個類。我們使用RedisTemplate操作Redis,這個里邊封裝好我們所需要的Redis的一些方法,用起來非常方便。這個類允許延遲任務做多有10W個,也是避免數(shù)據(jù)量過大對Redis造成影響。如果在線上使用的時候也需要考慮延遲任務的多少。太多幾百萬幾千萬的時候可能數(shù)據(jù)量非常大,我們需要計算Redis的空間是否夠。這個代碼也是非常的簡單,一個用于存放需要延遲的消息,采用offer的方法。另外一個是啟動一個線程, 如果消息時間到了,那么就將數(shù)據(jù)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 RedisTemplateString, String> redisTemplate;
    private String queueName;
    private BlockingQueueMessage> delayedQueue;

    public RedisListDelayedQueue(RedisTemplateString, String> redisTemplate, String queueName, BlockingQueueMessage> 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的代碼位置。我們也是生成了兩個消息,然后把消息放到隊列里邊,另外我們在啟動一個線程任務,用于將數(shù)據(jù)從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,后續(xù)所有用到的key和value都是String的
    @Autowired
    RedisTemplateString, 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";

        BlockingQueueMessage> 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";

        BlockingQueueMessage> 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);
                SetString> 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();
                }
            }
        });
    }

}

  我就不把運行結(jié)果寫出來了,感興趣的同學自己自行試驗。當然這個方法也是從內(nèi)存中拿出數(shù)據(jù),到時間之后放到Redis里邊,還是會存在程序啟動的時候,任務進行丟失。我們繼續(xù)看另外一種方法更好的進行這個問題的處理。

3.使用Redis的zSet實現(xiàn)分布式延遲隊列。

  我們需要再寫一個ZSet的隊列處理。下邊的offerMessage主要是把消息直接放入緩存中。采用Redis的ZSET的zadd方法。zadd(key, value, score) 即將key=value的數(shù)據(jù)賦予一個score, 放入緩存中。score就是計算出來延遲的毫秒數(shù)。

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 RedisTemplateString, String> redisTemplate;
    private String queueName;
    private BlockingQueueMessage> delayedQueue;

    public RedisZSetDelayedQueue(RedisTemplateString, String> redisTemplate, String queueName, BlockingQueueMessage> 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方法已經(jīng)寫好了測試的方法。/redis/zSetDelayedQueue,里邊主要使用ZSet的zRangeByScore(key, min, max)。主要是從score從0,當前時間的毫秒數(shù)獲取。取出數(shù)據(jù)后再采用removeRangeByScore,將數(shù)據(jù)刪除。這樣數(shù)據(jù)可以直接寫到Redis里邊,然后取出數(shù)據(jù)后直接處理。這種方法比前邊的方法稍微好一些,但是實際上還存在一些問題,因為依賴Redis,如果Redis內(nèi)存不足或者連不上的時候,系統(tǒng)將變得不可用。

4. 總結(jié)一下,另外還有哪些可以延遲隊列。

  上面的方法其實還是存在問題的,比如系統(tǒng)重啟的時候還是會造成任務的丟失。所以我們在生產(chǎn)上使用的時候,我們還需要將任務保存起來,比如放到數(shù)據(jù)庫和文件存儲系統(tǒng)將數(shù)據(jù)存儲起來,這樣做到double-check,雙重檢查,最終達到任務的99.999%能夠處理。

  其實還有很多東西可以實現(xiàn)延遲隊列。

  1) RabbitMQ就可以實現(xiàn)此功能。這個消息隊列可以把數(shù)據(jù)保存起來并且進行處理。

  2)Kafka也可以實現(xiàn)這個功能。

  3)Netty的HashedWheelTimer也可以實現(xiàn)這個功能。

最后放上我的代碼: https://github.com/stonehqs/delayQueue

到此這篇關于Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)的文章就介紹到這了,更多相關Redis延遲隊列和分布式延遲隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!

您可能感興趣的文章:
  • 基于Redis延遲隊列的實現(xiàn)代碼
  • SpringBoot集成Redisson實現(xiàn)延遲隊列的場景分析
  • php使用redis的有序集合zset實現(xiàn)延遲隊列應用示例

標簽:北京 朝陽 吉安 大慶 臺州 果洛 江蘇 楊凌

巨人網(wǎng)絡通訊聲明:本文標題《Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)》,本文關鍵詞  Redis,延遲,隊列,和,分布式,;如發(fā)現(xiàn)本文內(nèi)容存在版權(quán)問題,煩請?zhí)峁┫嚓P信息告之我們,我們將及時溝通與處理。本站內(nèi)容系統(tǒng)采集于網(wǎng)絡,涉及言論、版權(quán)與本站無關。
  • 相關文章
  • 下面列出與本文章《Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)》相關的同類信息!
  • 本頁收集關于Redis延遲隊列和分布式延遲隊列的簡答實現(xiàn)的相關信息資訊供網(wǎng)民參考!
  • 推薦文章