redis实现消息队列的几种方式
一、了解
众所周知，redis是我们日常开发过程中使用最多的非关系型数据库，也是消息中间件。实际上除了常用的rabbitmq、rocketmq、kafka消息队列（大家自己下去研究吧~模式都是通用的），我们也能使用redis实现消息队列。因为其他中间件可能更适用于大型/企业级项目，在咱们项目前期不需要这么多的数据，redis跟我们也是高度集成的。这里就简化了技术栈。
二、常用的几种使用redis实现的消息队列方式
1、List数据结构
Redis列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）一个列表最多可以包含 232 - 1 个元素 (4294967295, 每个列表超过40亿个元素)。
这里的列表大家可以想想为一个横着的通道，假设我现在往右边插入第一条数据，这个元素就会被放在最左边，接着再放入第二条数据，它就会在左边第二条，以此类推…插入了100条数据。 假设这个时候我要取出第一条，我就从最左边取就好。
这就变相实现了有序消息队列。具体实现大家自己研究
优点：操作方便，可以有序的取出自己插入的数据
缺点：不能进行实时消费，没有消费者
2、pub/sub 订阅消费模式
这就是传统的生产者->队列->消费者的模式。生产者的消息所有订阅者都能收到。
优点：实现了发布订阅模式，可以实时进行消费
缺点：没有消息持久化，在系统崩溃、宕机的时候；消息会丢失
3、sorted set有序集合Redis
有序集合和集合一样也是 string 类型元素的集合,且不允许重复的成员。不同的是每一个元素都会关联一个double分数，redis就是通过分数为集合中的成员进行从大到小的排列。
有序集合的成员是唯一的，但是score是可以重复的。
生成消息直接往s-set中插入数据，将score设置为接收到数据的13位时间戳；需要使用的时候再根据score大小有序取出来就行了。
看到这里是不是大家能想到，既然每条消息都带有时间，那我是不是可以顺手实现延迟队列。
这里只需要将score设置为 接受消息的时间戳+延迟时间 。我在使用的时候获取当天时间戳的数据，这样就实现了延迟消息队列。
优点：操作方便，可以实现延迟队列
缺点：不能实时进行消费
4、stream流 （redis5.0版本以上才有 重点讲）
Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。
Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：
每个stream流都有自己的名称，它是redis的key，也可以理解为队列名称。
Consumer Group ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。
stream常用命令
- XADD 定义stream流，写入消息体
XADD mystream * field1 A field2 B field3 C field4 D
mystream：自定义流名称
*：由redis生成流的id（也可以自定义，但是得保证自增唯一）
field1-A \field2-B\field3-C ：保存的消息体，key-value形式
-- 举例
redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
- XDEL 删除消息
> XADD mystream * a 1
1538561698944-0
> XADD mystream * b 2
1538561700640-0
> XADD mystream * c 3
1538561701744-0
> XDEL mystream 1538561700640-0
(integer) 1
127.0.0.1:6379> XRANGE mystream - +
1) 1) 1538561698944-0
2) 1) "a"
2) "1"
2) 1) 1538561701744-0
2) 1) "c"
2) "3"
- XRANGE 获取消息队列数据
XRANGE key start end [COUNT count]
key:strem流名称
start：开始值，- 表示最小值
end：结束值，+ 表示最大值
-- 举例：
redis> XRANGE mystream - + 2
从mystrem全部数据中取出两条数据
redis> XRANGE mystream + - 1
从mystream倒叙取一条数据
- XREVRANGE 自动过滤已删除的消息
redis> XADD writers * name Virginia surname Woolf
"1601372731458-0"
redis> XADD writers * name Jane surname Austen
"1601372731459-0"
redis> XADD writers * name Toni surname Morrison
"1601372731459-1"
redis> XADD writers * name Agatha surname Christie
"1601372731459-2"
redis> XADD writers * name Ngozi surname Adichie
"1601372731459-3"
redis> XLEN writers
(integer) 5
redis> XREVRANGE writers + - COUNT 1
1) 1) "1601372731459-3"
2) 1) "name"
2) "Ngozi"
3) "surname"
4) "Adichie"
redis>
- XREAD 阻塞或者非阻塞获取消息
# 从 Stream 头部读取两条消息
> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
1) 1) "mystream"
2) 1) 1) 1526984818136-0
2) 1) "duration"
2) "1532"
3) "event-id"
4) "5"
5) "user-id"
6) "7782813"
2) 1) 1526999352406-0
2) 1) "duration"
2) "812"
3) "event-id"
4) "9"
5) "user-id"
6) "388234"
2) 1) "writers"
2) 1) 1) 1526985676425-0
2) 1) "name"
2) "Virginia"
3) "surname"
4) "Woolf"
2) 1) 1526985685298-0
2) 1) "name"
2) "Jane"
3) "surname"
4) "Austen"
count ：数量
milliseconds ：可选，阻塞毫秒数，没有设置就是非阻塞模式
key ：队列名
id ：消息 ID
- XGROUP CREATE 创建消费者组
XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]
key ：队列名称，如果不存在就创建
groupname ：组名。
$ ： 表示从尾部开始消费，只接受新消息，当前 Stream 消息会全部忽略
从头开始消费:
XGROUP CREATE mystream consumer-group-name 0-0
从尾部开始消费:
XGROUP CREATE mystream consumer-group-name $
以上就是常用的steam流的命令，大家下来自己测试，练习。
三、springboot整合redis stream流
java中提供了连接redis的客户端，jedis和lettuce、redistemplate；RedisTemplate 是 Spring Data Redis 提供的一个高级抽象层，封装了 Jedis 或 Lettuce 等底层客户端。
它提供了丰富的功能，如序列化、事务支持、键过期等。这里主要讲主流的redistemplate整合，大家以后能直接使用。
实时消费
实时消费顾名思义，生产者发送消息，消费者立马进行消费逻辑处理。
- RedisStreamUtils工具类，方便后续进行stream操作没根据自己项目需求来定义
@Configuration
@SuppressWarnings("all")
public class RedisStreamUtils {
@Resource
private RedisTemplate<String, Object> redisTemplate;
/**
* 创建消费组
*
* @param streamKey 键名称
* @param group 组名称
* @return {@link String}
*/
public String createGroup(String streamKey, String group) {
return redisTemplate.opsForStream().createGroup(streamKey, group);
}
/**
* 获取消费者信息
*
* @param streamKey 键名称
* @param group 组名称
* @return {@link StreamInfo.XInfoConsumers}
*/
public StreamInfo.XInfoConsumers queryConsumers(String streamKey, String group) {
return redisTemplate.opsForStream().consumers(streamKey, group);
}
/**
* 查询组信息
*
* @param streamKey 键名称
* @return
*/
public StreamInfo.XInfoGroups queryGroups(String streamKey) {
return redisTemplate.opsForStream().groups(streamKey);
}
// 添加Map消息
public String addMap(String streamKey, Map<String, Object> value) {
return Objects.requireNonNull(redisTemplate.opsForStream().add(streamKey, value)).getValue();
}
// 读取消息
public List<MapRecord<String, Object, Object>> read(String streamKey) {
return redisTemplate.opsForStream().read(StreamOffset.fromStart(streamKey));
}
// 确认消费
public Long ack(String streamKey, String group, String... recordIds) {
return redisTemplate.opsForStream().acknowledge(streamKey, group, recordIds);
}
// 删除消息。当一个节点的所有消息都被删除，那么该节点会自动销毁
public Long del(String key, String... recordIds) {
return redisTemplate.opsForStream().delete(key, recordIds);
}
// 判断是否存在key
public boolean hasKey(String key) {
Boolean aBoolean = redisTemplate.hasKey(key);
return aBoolean != null && aBoolean;
}
}
- RedisConfig配置文件
@Configuration
@Slf4j
@RequiredArgsConstructor
public class RedisConfig {
private final RedisStreamUtils redisStreamUtil;
private final Environment environment;
//消费者处理消息配置
@Bean
public Subscription subscription(RedisConnectionFactory factory) {
AtomicInteger index = new AtomicInteger(1);
//获取系统处理器数量 创建线程池，开启守护线程
int processors = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(), r -> {
Thread thread = new Thread(r);
thread.setName("async-stream-consumer-" + index.getAndIncrement());
thread.setDaemon(true);
return thread;
});
//流消息监听容器参数设置
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
StreamMessageListenerContainer
.StreamMessageListenerContainerOptions
.builder()
// 一次最多获取多少条消息
.batchSize(5)
//执行线程池
.executor(executor)
//阻塞消息读取（延迟消息）
.pollTimeout(Duration.ofSeconds(1))
//异常处理
.errorHandler(throwable -> {
log.error("[MQ handler exception]", throwable);
throwable.printStackTrace();
})
.build();
//通过redis连接工厂，创建流消息监听容器
var listenerContainer = StreamMessageListenerContainer.create(factory, options);
//初始化流和消费者处理配置
//初始化流和消费者处理配置
Subscription subscription = initStreamAndConsumer(listenerContainer);
//开启监听容器
listenerContainer.start();
return subscription;
}
private Subscription initStreamAndConsumer(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer){
//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
//这一部分可以不用配置，可以根据自己的实际情况配置
//该key和group可根据需求自定义配置
String streamName = "mystream";
String groupname = "mygroup";
initStream(streamName, groupname);
// 手动ask消息
//消费者处理完消息之后，会进行确认；这里有一个pending状态会变成已处理
Subscription subscription = listenerContainer.receive(Consumer.from(groupname, "zhuyazhou"),
StreamOffset.create(streamName, ReadOffset.lastConsumed()), new RedisConsumer(redisStreamUtil));
// 自动ask消息
/* Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from(redisMqGroup.getName(), redisMqGroup.getConsumers()[0]),
StreamOffset.create(streamName, ReadOffset.lastConsumed()), new ReportReadMqListener());*/
//这一部分可以不用配置，可以根据自己的实际情况配置
//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
return subscription;
}
private void initStream(String key, String group) {
boolean hasKey = redisStreamUtil.hasKey(key);
if (!hasKey) {
Map<String, Object> map = new HashMap<>(1);
map.put("field", "value");
//创建主题
String result = redisStreamUtil.addMap(key, map);
//创建消费组
redisStreamUtil.createGroup(key, group);
//将初始化的值删除掉
redisStreamUtil.del(key, result);
log.info("stream:{}-group:{} initialize success", key, group);
}
}
}
大家这里可以想一想，这种写法是不是符合生产过程中的创建队列/消费者的逻辑，是不是不方便。能不能在我需要的时候直接调用方法去创建？？？假设现在我新增了一个业务需求，需要用不同的业务逻辑去处理，而且我希望定制不同的消费者应答模式，这个时候就需要一个通用方法去实现，这里我是这样做的。还是在工具类中
创建redis流消息监听容器
主要参数 ：定义线程池、一次最大获取消息数、超时重新获取、异常处理
@Bean
public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory factory) {
log.info("redis ip:{},port:{}",environment.getProperty("spring.data.redis.host"),environment.getProperty("spring.data.redis.port"));
AtomicInteger index = new AtomicInteger(1);
int processors = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(), r -> {
Thread thread = new Thread(r);
thread.setName("async-stream-consumer-" + index.getAndIncrement());
thread.setDaemon(true);
return thread;
});
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
StreamMessageListenerContainer
.StreamMessageListenerContainerOptions
.builder()
// 一次最多获取多少条消息
.batchSize(5)
.executor(executor)
.pollTimeout(Duration.ofSeconds(3))
.errorHandler(throwable -> {
log.error("[MQ handler exception]", throwable);
throwable.printStackTrace();
})
.build();
return StreamMessageListenerContainer.create(factory, options);
}
//业务需求调用此方法即可
public void addNewStreamAndSubscribe(String streamName, String groupName, String consumerId, StreamListener listener) {
initStream(streamName, groupName);
subscribeToStream(streamName, groupName, Consumer.from(groupName, consumerId), listener);
}
public void addNewStreamAndSubscribe(String streamName, String groupName, String consumerId, RedisConsumer listener,Map<String,Object> recodMap) {
initStream(streamName, groupName);
subscribeToStream(streamName, groupName, Consumer.from(groupName, consumerId), listener);
addMap(streamName, recodMap);
}
private void subscribeToStream(String streamName, String groupName, Consumer consumer, StreamListener listener) {
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = streamMessageListenerContainer(redisConnectionFactory);
Subscription subscription = container.receive(Consumer.from(groupName, consumer.getName()),
StreamOffset.create(streamName, ReadOffset.lastConsumed()), listener);
//开始消息容器监听
container.start();
log.info("Subscribed to stream: {} with group: {} and consumer: {}", streamName, groupName, consumer);
}
streamMessageListenerContainer中的 .batchSize(1) 设置需要着重说一下。意思是在消费者在监听到数据的时候，一次从redis中取出的多少条数据，假设我设置1，就意味着我的监听器会redis中取出1条未消费的数据，随后进入消费者逻辑，处理完毕之后返回；继续由监听器读取1条数据，在进入消费者逻辑；这个值设置得越小消息处理数据越快，但是也会增加redis链接的资源。
较大的 batchSize 可以减少与 Redis 服务器的交互次数，降低网络通信开销，提高处理效率。
较小的 batchSize 适用于需要低延迟处理的场景，但会增加网络通信开销和 CPU 使用率。
- RedisConsumer消费者
@Component("RedisConsumer")
@RequiredArgsConstructor
@Slf4j
public class RedisConsumer implements StreamListener<String, MapRecord<String,String,String>> {
private final RedisStreamUtils redisStreamUtils;
@Override
public void onMessage(MapRecord<String, String, String> message) {
try {
log.info("RedisConsumer1获取到了消息：{}",message);
String streamKey = message.getStream();
RecordId recordId = message.getId();
Map<String, String> value = message.getValue();
//获取这个流下 所有的消费者组
StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);
//处理逻辑
//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
log.info("【streamKey】= {}，【recordId】= {},【msg】= {}",streamKey,recordId, value);
//手动确认ack消息，并删除已处理的消息
//我这里使用手动
xInfoGroups.forEach(xInfoGroup -> redisStreamUtils.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));
//自动确认消息 ---------自己下来研究
//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
//根据业务场景来看是否需要删除消息
// redisStreamUtils.del(streamKey, recordId.getValue());
} catch (Exception e) {
throw new ServiceException("消费异常");
}
}
}
- RedisConsumer2消费者
@Component("RedisConsumer2")
@RequiredArgsConstructor
@Slf4j
public class RedisConsumer2 implements StreamListener<String, MapRecord<String,String,String>> {
private final RedisStreamUtils redisStreamUtils;
@Override
public void onMessage(MapRecord<String, String, String> message) {
try {
log.info("RedisConsumer2获取到了消息：{}",message);
String streamKey = message.getStream();
RecordId recordId = message.getId();
Map<String, String> value = message.getValue();
//获取这个流下 所有的消费者组
StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);
//处理逻辑
//↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
log.info("【streamKey】= {}，【recordId】= {},【msg】= {}",streamKey,recordId, value);
//↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
//手动确认ack消息，并删除已处理的消息
xInfoGroups.forEach(xInfoGroup -> redisStreamUtils.ack(streamKey, xInfoGroup.groupName(), recordId.getValue()));
// redisStreamUtils.del(streamKey, recordId.getValue());
} catch (Exception e) {
throw new ServiceException("消费异常");
}
}
}
- RedisStreamcontroller模拟测试
@RequestMapping(value = "/redisStream")
@RestController
@RequiredArgsConstructor
@Slf4j
@SuppressWarnings("all")
public class RedisStreamController {
private final RedisStreamUtils redisStreamUtils;
private final RedisConsumer redisConsumer;
private final RedisTemplate redisTemplate;
private final ApplicationContext applicationContext;
@GetMapping(value = "/addNewStreamAndSubscribe")
public ResultVO addNewStreamAndSubscribe(@RequestParam("streamKey") String streamKey,
@RequestParam("groupName") String groupName,
@RequestParam("consumer")String consumer,
@RequestParam("consumerClass") String consumerClass){
try {
// 获取实现类的实例
StreamListener consumerInstance = (StreamListener) applicationContext.getBean(consumerClass);
redisStreamUtils.addNewStreamAndSubscribe(streamKey, groupName, consumer,consumerInstance );
} catch (Exception e) {
throw new RuntimeException(e);
}
return ResultVO.success();
}
@GetMapping(value = "/addMap")
public ResultVO addMap(@RequestParam("streamKey") String streamKey,
@RequestParam("key")String key,
@RequestParam("value")String value) {
HashMap<String, Object> objectObjectHashMap = new HashMap<>();
objectObjectHashMap.put(key,value);
redisStreamUtils.addMap(streamKey,objectObjectHashMap);
return ResultVO.success();
}
@GetMapping(value = "/getGroup")
public ResultVO getGroup(@RequestParam("streamKey") String streamKey,
@RequestParam("groupName") String groupName) {
boolean b = redisStreamUtils.hasKey(streamKey);
if(b){
StreamInfo.XInfoGroups xInfoGroups = redisStreamUtils.queryGroups(streamKey);
List<Object> list = new ArrayList<>();
for (StreamInfo.XInfoGroup xInfoGroup : xInfoGroups) {
StreamInfo.XInfoConsumers xInfoConsumers = null;
if(StrUtil.isNotEmpty(groupName)){
xInfoConsumers = redisStreamUtils.queryConsumers(streamKey, groupName);
for (StreamInfo.XInfoConsumer xInfoConsumer : xInfoConsumers) {
log.info("group:{},pending:{},consumerCount:{},consumerName:{},lastDeliveryId:{}"
,xInfoGroup.groupName(),xInfoGroup.pendingCount(),xInfoGroup.consumerCount(),xInfoConsumer.consumerName(),xInfoGroup.lastDeliveredId());
}
}
}
}else{
log.info("streamKey不存在：{}",streamKey);
return ResultVO.error("streamKey不存在");
}
return ResultVO.success();
}
@GetMapping(value = "/delStream")
public ResultVO delStream(@RequestParam("streamKey") String streamKey){
redisTemplate.delete(streamKey);
return ResultVO.success();
}
@GetMapping(value = "/readMsg")
public ResultVO readMsg(@RequestParam("streamKey") String streamKey,
@RequestParam("groupName") String groupName,
@RequestParam("consumer") String consumer){
// 读取消息，每次读取最多 5 条
List read = redisTemplate.opsForStream().read(
Consumer.from(groupName, consumer),
StreamReadOptions.empty().count(10).block(Duration.ofSeconds(1)),
StreamOffset.create(streamKey, ReadOffset.lastConsumed())
);
return ResultVO.success(JSON.toJSONString(read));
}
项目启动
调用/addNewStreamAndSubscribe接口
- 创建流、监听容器
- 消费者绑定流+消费者逻辑处理类
- 接收生产者消息方式（最新、偏移量）
- 开启消息容器监听
调用/addMap接口，发送消息
如果只有一个消费者，那么当消费者出现异常的时候，直到服务恢复，会从上一次消费的数据开始进行消费。
假设现在消费者组有两个消费者，都绑定了同一个消息流，这个时候发送消息就是轮询访问。
RedisConsumer1获取到了消息
RedisConsumer2获取到了消息
RedisConsumer1获取到了消息
RedisConsumer2获取到了消息
…
如果consumer1出现了异常，这个时候consumer2会正常消费所有的数据。
stream本身就支持持久化数据，也是dbs和aof两种。不用担心数据丢失。