网站还没建设好可以备案吗,杭州网络公司做网站报价,南京优化网站建设,苏州网站建设点一点背景
看到消息队列#xff0c;我们肯定会想到各种MQ#xff0c;比如#xff1a;RabbitMQ#xff0c;acivityMQ、RocketMQ、Kafka等。
但是#xff0c;当我们需要使用消息中间件的时候#xff0c;并非每次都需要非常专业的消息中间件#xff0c;假如我们只有一个消息队… 背景
看到消息队列我们肯定会想到各种MQ比如RabbitMQacivityMQ、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 BootRedis实现
添加application.properties内容
# Redis数据库索引默认为0
spring.redis.database0
# Redis服务器地址
spring.redis.host127.0.0.1
# Redis服务器连接端口
spring.redis.port6379
# Redis服务器连接密码默认为空
spring.redis.password
# 连接池最大连接数使用负值表示没有限制
spring.redis.jedis.pool.max-active20
# 连接池最大阻塞等待时间使用负值表示没有限制
spring.redis.jedis.pool.max-wait-1
# 连接池中的最大空闲连接
spring.redis.jedis.pool.max-idle10
# 连接池中的最小空闲连接
spring.redis.jedis.pool.min-idle0
# 连接超时时间毫秒
spring.redis.timeout1000pom.xml中添加
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-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 {Autowiredprivate RedisConnectionFactory redisConnectionFactory;Beanpublic RedisTemplateString, Object redisTemplate() {RedisTemplateString, Object template new RedisTemplateString, 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;Resourceprivate RedisTemplate redisTemplate;Overridepublic void produce(String string) {//生产者把消息丢到消息队列中redisTemplate.opsForList().leftPush(MESSAGE_KEY, string);}Overridepublic 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 {Resourceprivate 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 {Resourceprivate 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*1000long 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();Resourceprivate DelayingQueueService delayingQueueService;private Logger log LoggerFactory.getLogger(this.getClass());/*** 定时消费队列中的数据* zset会对score进行排序 让最早消费的数据位于最前* 拿最前的数据跟当前时间比较 时间到了则消费*/Scheduled(cron */1 * * * * *)public void consumer() throws JsonProcessingException {ListMessage 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 { Resourceprivate 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来做消息队列。但如果对消息可靠性有高度要求的话 建议从上面的其他方案中选一个相对合适的来实现。
