文章目录
- 基本模式
- 生产者消费者
- 原理&模型
- redis实现
- java实现
- 发布者订阅者
- 原理&模型
- redis实现
- java实现
- stream模式
- 原理&模型
- 工作原理
- redis实现
- Java实现
- 选型
- 外传
基本模式
采用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以上
外传
😜 原创不易，如若本文能够帮助到您的同学
🎉 支持我：关注我+点赞👍+收藏⭐️
📝 留言：探讨问题，看到立马回复
💬 格言：己所不欲勿施于人 扬帆起航、游历人生、永不言弃！🔥
