田哥手把手教你用 Redis 做延迟消息队列

发表于 3年以前 | 总阅读数：1178 次

背景

看到消息队列，我们肯定会想到各种MQ，比如：RabbitMQ，acivityMQ、RocketMQ、Kafka等。

但是，当我们需要使用消息中间件的时候，并非每次都需要非常专业的消息中间件，假如我们只有一个消息队列，只有一个消费者，那就没有必要去使用上面这些专业的消息中间件，这种情况我们可以考虑使用 Redis 来做消息队列。

延迟消息队列使用场景

我们打车，在规定时间内，没有车主接单，那么平台就会推送消息给你，提示暂时没有车主接单。
网上支付场景，下单了，如果没有在规定时间付款，平台通常会发消息提示订单在有效期内没有支付完成，此订单自动取消之类的信息。
我们买东西，如果在一定时间内，没有对该订单进行评分，这时候平台会给一个默认分数给此订单。

.....

Redis如何实现消息队列？

大家都知道，Redis的五种数据类型，其中有一种类型是list。并且提供了相应的进入list的命令lpush和rpush ，以及弹出list的命令lpop和rpop。

这里我们就可以把List理解为一个消息队列，并且lpush和rpush操作称之为进入队列，同时，lpop和rpop称之为消息出队列。

命令

lpush

将一个或多个值 value 插入到列表 key 的表头
如果有多个 value 值，那么各个 value 值按从左到右的顺序依次插入到表头：比如说，对空列表 mylist 执行命令 LPUSH mylist a b c ，列表的值将是 c b a ，
这等同于原子性地执行 LPUSH mylist a 、 LPUSH mylist b 和 LPUSH mylist c 三个命令。
如果 key 不存在，一个空列表会被创建并执行 LPUSH 操作。
当 key 存在但不是列表类型时，返回一个错误。

使用案例：

rpush

将一个或多个值 value 插入到列表 key 的表尾(最右边)。
如果有多个 value 值，那么各个 value 值按从左到右的顺序依次插入到表尾：比如对一个空列表 mylist 执行 RPUSH mylist a b c ，得出的结果列表为 a b c ，等同于执行命令 RPUSH mylist a 、 RPUSH mylist b 、 RPUSH mylist c 。
如果 key 不存在，一个空列表会被创建并执行 RPUSH操作。
当 key 存在但不是列表类型时，返回一个错误。

使用案例：

lpop

使用方式:lpop key。移除并返回列表 key 的头元素。如果key不存在，返回nil。

使用案例：

rpop

使用方式：rpop key,移除并返回列表 key 的尾元素。当 key 不存在时，返回 nil 。

使用案例：

以上四个命令是不是相当的简单呢，这里说一下lrange命令。

lrange

返回列表 key 中指定区间内的元素，区间以偏移量 start 和 stop 指定。
下标(index)参数 start 和 stop 都以 0 为底，也就是说，以 0 表示列表的第一个元素，以 1 表示列表的第二个元素，以此类推。
你也可以使用负数下标，以 -1 表示列表的最后一个元素， -2 表示列表的倒数第二个元素，以此类推。

Redis实现消息队列

使用Spring Boot+Redis实现：

添加application.properties内容：

# Redis数据库索引（默认为0）
spring.redis.database=0
# Redis服务器地址
spring.redis.host=127.0.0.1
# Redis服务器连接端口
spring.redis.port=6379
# Redis服务器连接密码（默认为空）
spring.redis.password=
# 连接池最大连接数（使用负值表示没有限制）
spring.redis.jedis.pool.max-active=20
# 连接池最大阻塞等待时间（使用负值表示没有限制）
spring.redis.jedis.pool.max-wait=-1
# 连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle=10
# 连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle=0
# 连接超时时间（毫秒）
spring.redis.timeout=1000

pom.xml中添加：

<dependency>
   <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
 </dependency>

创建一个RedisConfig类，对RedisTemplate做一些序列化的设置：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class RedisConfig {

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
        template.setConnectionFactory(redisConnectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        template.afterPropertiesSet();
        return template;
    }
}

创建RedisMQServicehe RedisMQServiceImpl

public interface RedisMQService {
    void produce(String string);
    void consume();
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Service
public class RedisMQServiceImpl implements RedisMQService {
    private static Logger log = LoggerFactory.getLogger(RedisMQServiceImpl.class);

    private static final String MESSAGE_KEY = "message:queue";

    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void produce(String string) {
        //生产者把消息丢到消息队列中
        redisTemplate.opsForList().leftPush(MESSAGE_KEY, string);
    }

    @Override
    public void consume() {
        String string = (String) redisTemplate.opsForList().rightPop(MESSAGE_KEY);
        //消费方拿到消息后进行业务处理
        log.info("consume : {}", string);
    }
}

创建一个ccontroller

import com.tian.user.mq.RedisMQService;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
public class RedisMQController {
    @Resource
    private RedisMQService redisMQService;

    @PostMapping("/produce")
    public String produce() {
        String[] names = {"java后端技术全栈", "老田", "面试专栏"};
        for (String name : names) {
            redisMQService.produce(name);
        }
        return "ok";
    }

    @PostMapping("/consume")
    public void consume() {
        int i = 0;
        while (i < 3) {
            redisMQService.consume();
            i++;
        }
    }
}

启动项目，访问：http://localhost:8080/produce生产者把三个消息丢到消息队列中。

在访问：http://localhost:8080/consume从消息队列中取出消息，然后就可以拿着消息继续做相关业务了。

后台输出：

到此，使用Redis实现消息队列就成功了。

但是，搞了半天只是使用Redis实现了消息队列，那延迟呢？

上面并没有提到延迟队列的实现方式，下面我们来看看Redis中是如何实现此功能的。

Redis实现延迟消息队列的相关命令

延迟队列可以通过 zset 来实现，因为 zset 中有一个 score，我们可以把时间作为 score，将 value 存到 redis 中，然后通过轮询的方式，去不断的读取消息出来

整体思路

1.消息体设置有效期，设置好score，然后放入zset中

2.通过排名拉取消息

3.有效期到了，就把当前消息从zset中移除

我们来看看，zset有哪些命令：

可以通过网址：http://doc.redisfans.com/set/index.html 获取中文版Redis命令。

其实Redis实现延迟队列，只需要zset的三个命令即可。下面先来熟悉这三个命令。

zadd命令

使用方式：ZADD key score member [[score member][score member] ...]

将一个或多个 member 元素及其 score 值加入到有序集 key 当中。
如果某个 member 已经是有序集的成员，那么更新这个 member 的 score 值，并通过重新插入这个 member 元素，来保证该 member 在正确的位置上。
score 值可以是整数值或双精度浮点数。
如果 key 不存在，则创建一个空的有序集并执行 ZADD 操作。
当 key 存在但不是有序集类型时，返回一个错误。
在 Redis 2.4 版本以前， ZADD 每次只能添加一个元素。

使用案例：

ZRANGEBYSCORE命令

使用方式：ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]

1.返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。

2.具有相同 score 值的成员按字典序(lexicographical order)来排列(该属性是有序集提供的，不需要额外的计算)。

3.可选的 LIMIT 参数指定返回结果的数量及区间(就像SQL中的 SELECT LIMIT offset, count )，注意当 offset 很大时，定位 offset 的操作可能需要遍历整个有序集，此过程最坏复杂度为 O(N) 时间。

4.可选的 WITHSCORES 参数决定结果集是单单返回有序集的成员，还是将有序集成员及其 score 值一起返回。

注意：该选项自 Redis 2.0 版本起可用。

区间及无限

min 和 max 可以是 -inf 和 +inf ，这样一来，你就可以在不知道有序集的最低和最高 score 值的情况下，使用 ZRANGEBYSCORE这类命令。

默认情况下，区间的取值使用闭区间(小于等于或大于等于)，你也可以通过给参数前增加 ( 符号来使用可选的开区间 (小于或大于)。

举个例子：

ZRANGEBYSCORE zset (1 5

返回所有符合条件 1 < score <= 5 的成员，而

ZRANGEBYSCORE zset (5 (10

则返回所有符合条件 5 < score < 10 的成员。

使用案例

ZREM命令

使用方式：ZREM key member [member ...]

移除有序集 key 中的一个或多个成员，不存在的成员将被忽略。

当 key 存在但不是有序集类型时，返回一个错误。

使用案例：

延迟队列可以通过Zset(有序列表实现)，Zset类似于java中SortedSet和HashMap的结合体，它是一个Set结构，保证了内部value值的唯一，同时他还可以给每个value设置一个score作为排序权重，Redis会根据score自动排序，我们每次拿到的就是最先需要被消费的消息，利用这个特性我们可以很好实现延迟队列。

java代码实现

创建一个消息实体类：

import java.time.LocalDateTime;

public class Message {
    /**
     * 消息唯一标识
     */
    private String id;
    /**
     * 消息渠道 如 订单 支付 代表不同业务类型
     * 为消费时不同类去处理
     */
    private String channel;
    /**
     * 具体消息 json
     */
    private String body;

    /**
     * 延时时间 被消费时间  取当前时间戳+延迟时间
     */
    private Long delayTime;

    /**
     * 创建时间
     */
    private LocalDateTime createTime;
    // set get 省略
}

生产者方代码：

import com.tian.user.dto.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.UUID;

@Component
public class MessageProvider {

    @Resource
    private DelayingQueueService delayingQueueService;

    private static String USER_CHANNEL = "USER_CHANNEL";

    /**
     * 发送消息
     *
     * @param messageContent
     */
    public void sendMessage(String messageContent, long delay) {
        try {
            if (messageContent != null) {
                String seqId = UUID.randomUUID().toString();
                Message message = new Message();
                //时间戳默认为毫秒 延迟5s即为 5*1000
                long time = System.currentTimeMillis();
                LocalDateTime dateTime = Instant.ofEpochMilli(time).atZone(ZoneOffset.ofHours(8)).toLocalDateTime();
                message.setDelayTime(time + (delay * 1000));
                message.setCreateTime(dateTime);
                message.setBody(messageContent);
                message.setId(seqId);
                message.setChannel(USER_CHANNEL);
                delayingQueueService.push(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消费方代码：

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.tian.user.dto.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;

@Component
public class MessageConsumer {
    private static ObjectMapper mapper = Jackson2ObjectMapperBuilder.json().build();
    @Resource
    private DelayingQueueService delayingQueueService;
    private Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 定时消费队列中的数据
     * zset会对score进行排序 让最早消费的数据位于最前
     * 拿最前的数据跟当前时间比较 时间到了则消费
     */
    @Scheduled(cron = "*/1 * * * * *")
    public void consumer() throws JsonProcessingException {
        List<Message> msgList = delayingQueueService.pull();
        if (null != msgList) {
            long current = System.currentTimeMillis();
            msgList.stream().forEach(msg -> {
                // 已超时的消息拿出来消费
                if (current >= msg.getDelayTime()) {
                    try {
                        log.info("消费消息：{}:消息创建时间：{},消费时间：{}", mapper.writeValueAsString(msg), msg.getCreateTime(), LocalDateTime.now());
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                    //移除消息
                    try {
                        delayingQueueService.remove(msg);
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

controller中的代码(或自写一个test类)：

@RestController
public class RedisMQController { 
    @Resource
    private MessageProvider messageProvider; 

    @PostMapping("/delay/produce")
    public String produce() { 
        //延迟20秒
        messageProvider.sendMessage("同时发送消息1", 20);
        messageProvider.sendMessage("同时发送消息2", 20);
        return "ok";
    }
}

启动项目，访问：http://localhost:8080/delay/produce

后台消费者消费消息：