实现简单延迟队列和分布式延迟队列

时间  2020-04-19
标签 实现 简单 延迟 队列 分布式 栏目 系统架构 繁體版
原文   原文链接

　　在咱们的工做中，不少地方使用延迟队列，好比订单到期没有付款取消订单，制订一个提醒的任务等都须要延迟队列，那么咱们须要实现延迟队列。咱们本文的梗概以下，同窗们能够选择性阅读。java

　　1. 实现一个简单的延迟队列。

　　2.使用Redis的list实现分布式延迟队列。

　　3.使用Redis的zSet实现分布式延迟队列。

　　4. 总结一下，另外还有哪些能够延迟队列。

 

1.  实现一个简单的延迟队列。git

　　咱们知道目前JAVA能够有DelayedQueue,咱们首先开一个DelayQueue的结构类图。DelayQueue实现了Delay、BlockingQueue接口。也就是DelayQueue是一种阻塞队列。github

　　

 

　　咱们在看一下Delay的类图。Delayed接口也实现了Comparable接口，也就是咱们使用Delayed的时候须要实现CompareTo方法。由于队列中的数据须要排一下前后，根据咱们本身的实现。Delayed接口里边有一个方法就是getDelay方法，用于获取延迟时间，判断是否时间已经到了延迟的时间，若是到了延迟的时间就能够从队列里边获取了。web

 

　　咱们建立一个Message类，实现了Delayed接口，咱们主要把getDelay和compareTo进行实现。在Message的构造方法的地方传入延迟的时间，单位是毫秒，计算好触发时间fireTime。同时按照延迟时间的升序进行排序。我重写了里边的toString方法，用于将Message按照我写的方法进行输出。redis

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秒，时间到了以后会将接取出而且打印。输出的结果以下，正是咱们所指望的。spring

1587218430786:start
1587218435789:hello
1587218437793:world

　　这个方法实现起来真的很是简单。可是缺点也是很明显的，就是数据在内存里边，数据比较容易丢失。那么咱们须要采用Redis实现分布式的任务处理。docker

　　2. 使用Redis的list实现分布式延迟队列。数据库

　　本地须要安装一个Redis，我本身是使用Docker构建一个Redis，很是快速，命令也没多少。咱们直接启动Redis而且暴露6379端口。进入以后直接使用客户端命令便可查看和调试数据。apache

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　　

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程