Redis 消息队列和发布订阅

基本模式

采用redis
三种方案：
● 生产者消费者：一个消息只能有一个消费者
● 发布者订阅者：一个消息可以被多个消费者收到
● stream模式：实现队列和广播模式

生产者消费者

原理&模型

Producer调用redis的lpush往特定key里放消息，Consumer调用brpop去不断监听key。

1、利用redis的链表，存储数据，实现队列模式
2、生产者左进，消费者右出
3、消费者，线程拉取数据模式

redis实现

10.255.20.232:1>lpush test_mq 'lwd'
"1"
10.255.20.232:1>lpush test_mq 'lwd2'
"2"
10.255.20.232:1>lpush test_mq 'lwd3'
"3"
10.255.20.232:1>lrange test_mq 0 10
1) "lwd3"
2) "lwd2"
3) "lwd"

10.255.20.232:1>brpop test_mq 0
1) "test_mq"
2) "lwd"

10.255.20.232:1>lrange test_mq 0 10
1) "lwd3"
2) "lwd2"

java实现

//producer
String key="demo:mq:test";
String msg="hello world";
redisDao.lpush(key,msg);


//consumer,超时时间因情况而定
String key="demo:mq:test";
while(true){
   List<String> msgs=redisTemp.brpop(BLOCK_TIMEOUT,listKey);
   if(msgs==null)continue;
   String jobMsg=msgs.get(1);
   processMsg(jobMsg);
}

发布者订阅者

redis从2.0版本开始支持pub/sub
Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。
类似设计模式中的观察者模式。

redis可以作为pub/sub的服务端。订阅得通过subscribe和psubscribe命令向redis server订阅消息类型，redis将消息类型称为channel。

当发布者通过publish命令向redis server发送特定消息时，访问该消息类型全部client都会收到此消息。
一个client可以订阅多个channel，也可以向多个channel发送消息。

原理&模型

1、利用redis订阅模式，redis不存储数据
2、实时获取channel数据，不能获取历史数据(redis不存)
3、发布者发布不同topic，有订阅就获取，没有也不会保留

redis实现

# sub1
10.255.20.232:1>SUBSCRIBE test_pub_sub
Switch to Pub/Sub mode. Close console tab to stop listen for messages.
1) "subscribe"
2) "test_pub_sub"
3) "1"

1) "message"
2) "test_pub_sub"
3) "lwd1"

1) "message"
2) "test_pub_sub"
3) "lwd2"

-- sub2
10.255.20.232:1>SUBSCRIBE test_pub_sub
Switch to Pub/Sub mode. Close console tab to stop listen for messages.
1) "subscribe"
2) "test_pub_sub"
3) "1"

1) "message"
2) "test_pub_sub"
3) "lwd1"

1) "message"
2) "test_pub_sub"
3) "lwd2"

-- pub1
10.255.20.232:1>PUBLISH test_pub_sub 'lwd1'
"2"
10.255.20.232:1>PUBLISH test_pub_sub 'lwd2'
"2"

java实现

待实现

stream模式

Stream是Redis5.0添加的一种数据结构，是Redis对消息队列的实现。Redis 对 Streams 的采用受到 Apache Kafka 的启发，并保留了 Redis 的简单性及其极其有趣和有效的用法。

Redis Stream 主要用于消息队列（MQ，Message Queue），Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。
简单来说发布订阅 (pub/sub) 可以分发消息，但无法记录历史消息。
而 Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。

原理&模型

工作原理

1、通过stream数据类型做存储
2、通过group模式，实现队列，同一个组对应一个stream只能消费一次
3、纯mq模式
4、功能上有ack回执
5、多个group，对应一个stream，不同group中的消费者都可以消费到此stream
6、能够持久化

redis实现

-- server 增加stream
10.255.20.232:1>XADD student_stream * name Tom age 8
"1705387359901-0"
10.255.20.232:1>XADD student_stream * name lwd age 18
"1705387391303-0"
-- 创建组student_group 并绑定stream 并指定从0开始拿数据
10.255.20.232:1>XGROUP CREATE student_stream student_group 0
"OK"

-- client1 
-- consumer_1监听组中的stream，每次拿1条数九，阻塞100s
10.255.20.232:1>XREADGROUP GROUP student_group consumer_1 COUNT 1 BLOCK 100000 STREAMS student_stream >
1) 1) "student_stream"
   2) 1) 1) "1705309636354-0"
         2) 1) "name"
            2) "lwd"
            3) "age"
            4) "33"
-- 查看somestream中所有元素
XRANGE student_stream - +

-- 查看somestream中所有元素
XRANGE student_stream - +

-- 返回消费组student_group有多少未确认条数，起始ID，结束ID，消费者及其未确认条数
XPENDING student_stream student_group

10.255.20.232:1>XPENDING student_stream student_group
1) "1"
2) "1705309636354-0"
3) "1705309636354-0"
4) 1) 1) "consumer_1"
      2) "1"

-- ack 回执确认读取
10.255.20.232:1>XPENDING student_stream student_group
1) "1"
2) "1705309636354-0"
3) "1705309636354-0"
4) 1) 1) "consumer_1"
      2) "1"



10.255.20.232:1>XACK student_stream student_group 1705309636354-0
"1"
-- 再查看已读未确认消息
10.255.20.232:1>XPENDING student_stream student_group
1) "0"
2) null
3) null
4) null

Java实现

需要建立mq的管理机制
topic group可管理配置

选型

轻量级mq，可使用redis stream 消息模式，实现队列和发布订阅，但，需要redis版本5.0以上

外传

😜 原创不易，如若本文能够帮助到您的同学
🎉 支持我：关注我+点赞👍+收藏??
📝 留言：探讨问题，看到立马回复
💬 格言：己所不欲勿施于人 扬帆起航、游历人生、永不言弃！🔥