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){
   ListString> 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_strea服务器托管网m * 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以上

外传

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

服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net

相关推荐: 面试官：Redis如何保证高可用？

Redis 高可用（High Availability，HA）是指 Redis 通过一系列技术手段确保在面临故障的情况下也能持续提供服务的能力。 Redis 作为一个内存数据库，其数据通常存储在内存中，一旦发生故障，可能导致数据丢失或服务中断，所以，为了保证 …