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

千里之行,始于足下——老子

背景

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

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

延迟消息队列使用场景

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

.....

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

后台消费者消费消息:

从输出日志中,可以看出,已经实现了延迟的功能。

自此,Redis实现延迟队列的功能就完成了。

实现延迟队列的其他方案

「RabbitMQ」 :利用 RabbitMQ 做延时队列是比较常见的一种方式,而实际上RabbitMQ自身并没有直接支持提供延迟队列功能,而是通过 RabbitMQ 消息队列的 TTL和 DXL这两个属性间接实现的。

「RocketMQ」 :RocketMQ 发送延时消息时先把消息按照延迟时间段发送到指定的队列中(rocketmq把每种延迟时间段的消息都存放到同一个队列中),然后通过一个定时器进行轮训这些队列,查看消息是否到期,如果到期就把这个消息发送到指定topic的队列中。

「Kafka」 :Kafka支持延时生产、延时拉取、延时删除等,其基于时间轮和 JDK 的 DelayQueue 实现 。

「ActiveMQ」 :需要延迟的消息会先存储在JobStore中,通过异步线程任务JobScheduler将到达投递时间的消息投递到相应队列上 。

消息队列对比

总结

如果项目中仅仅是使用个别不是很重要的业务功能,可以使用Redis来做消息队列。但如果对消息可靠性有高度要求的话 ,建议从上面的其他方案中选一个相对合适的来实现。

参考:http://ii081.cn/wvN4B http://ccx4.cn/UxGIJ
(0)

相关推荐