平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“Redis怎么做延迟消息队列？zset实现方法延迟队列的原理和代码示例”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。
背景
实际处理时，看到消息队列，我们肯定会想到各种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++;
}
}
}
启动项目，访问：（链接已移除）生产者把三个消息丢到消息队列中。
在访问：（链接已移除）从消息队列中取出消息，随后就能够拿着消息继续做相关业务了。
后台输出：
到此，采用Redis实现消息队列就成功了。
但是，搞了半天只是采用Redis实现 了消息队列，那延迟呢？
理解这一步时，上面同时没有提到延迟队列的实现方法，下面我们来看看Redis中是如何实现此功能的。
Redis实现延迟消息队列的相关命令
结合项目来看，延迟队列能够借助 zset 来实现，因为 zset 中有一个 score，我们能够把时间作为 score，将 value 存到 redis 中，随后借助轮询的方式，去不断的读取消息出来
整体思路
1.消息体设置有效期，设置好score，随后放入zset中
2.借助排名拉取消息
3.有效期到了，就把当前消息从zset中移除
我们来看看，zset有哪些命令：
能够借助网址：（链接已移除） 拿到中文版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";
}
}
启动项目，访问：（链接已移除）
后台消费者消费消息：
从输出日志中，能够看出，已经实现了延迟的功能。
自此，Redis实现延迟队列的功能就完成了。
实现延迟队列的其他方案
- 「RabbitMQ」：借助 RabbitMQ 做延时队列是比较常用的一种方式，而实际上RabbitMQ自身同时没有直接兼容提供延迟队列功能，而是借助 RabbitMQ 消息队列的 TTL和 DXL这两个属性间接实现的。
- 「RocketMQ」：RocketMQ 发送延时消息时先把消息按照延迟时间段发送到指定的队列中(rocketmq把每种延迟时间段的消息都存放到同一个队列中)，随后借助一个定时器进行轮训这些队列，查看消息是否到期，如果到期就把这个消息发送到指定topic的队列中。
- 「Kafka」：Kafka兼容延时生产、延时拉取、延时删除等，其基于时间轮和 JDK 的 DelayQueue 实现。
- 「ActiveMQ」：需延迟的消息会先存储在JobStore中，借助异步线程任务JobScheduler将到达投递时间的消息投递到相应队列上。
消息队列对比
总结
在这个场景下，若项目中仅仅是采用个别不是很重要的业务功能，能够采用Redis来做消息队列。但如果对消息可靠性有高度要求的话，建议从上面的其他方案中选一个相对合适的来实现。
在这个场景下，总的来说，Redis延迟消息队列适合结合实际项目边做边理解。先抓住核心思路，再逐步补上细节和边界处理，最后效果会更稳定，也更容易复用。