redis pub sub 生产者可以当消费者吗

Posted 2023-05-06

参考技术A 在pub/sub中消息发布者不需要独占一个Redis的链接，而消费者则需要单独占用一个Redis的链接，在Java中便不得独立出分出一个线程来处理消费者。这种场景一般对应这多个消费者，此时则有着过高的资源消耗。

对于如上的几种不足，如果在项目中需要考虑的话可以使用JMS来实现该功能。JMS提供了消息的持久化/耐久性等各种企业级的特性。如果依然想使用Redis来实现并做一些数据的持久化操作，则可以根据JMS的特性来通过Redis模拟出来. 参考技术B 少量数据存储，高速读写访问。此类产品通过数据全部in-momery 的方式来保证高速访问，同时提供数据落地的功能，实际这正是Redis最主要的适用场景。

Redis pub/sub

Redis 中的pub/sub是指消息的发布订阅,是用来解耦系统的,以消息生产者和消息消费者的角色来定义两个系统.

本节主要介绍常用操作命令和Redis提供的两种通道.

一.操作命令

1. 消息订阅

subscribe channel [channel ...]

127.0.0.1:6379> subscribe topic1 topic2reading messages... (press ctrl-c to quit)1) "subscribe"2) "topic1"3) (integer) 11) "subscribe"2) "topic2"3) (integer) 2

2. 消息发布

publish channel message

客户端2发布信息

127.0.0.1:6379> publish topic1 message1(integer) 1

客户端1收到信息

1) "message"2) "topic1"3) "message1"

3. 系统channel订阅查询

pubsub channels

客户端2查询订阅通道

127.0.0.1:6379> pubsub channels1) "topic1"2) "topic2"

4. 查询每个通道的订阅数

pubsub numsub 

127.0.0.1:6379> pubsub numsub topic1 topic21) "topic1"2) (integer) 03) "topic2"4) (integer) 1

5. 按正则模式消息订阅

psubscribe pattern [pattern ...]

客户端1:

127.0.0.1:6379> psubscribe topic*reading messages... (press ctrl-c to quit)1) "psubscribe"2) "topic*"3) (integer) 1

客户端2:

127.0.0.1:6379> publish topic1 message1(integer) 2

客户端1收到信息

1) "pmessage"2) "topic*"3) "topic1"4) "message1"

6. 使用 psubscribe命令执行的订阅数

pubsub numpat

127.0.0.1:6379> pubsub numpat(integer) 1

7. 取消订阅

unsubscribe [channel [channel ...]]

二. 两种订阅通道

pub/sub api中提供的psubscribe和subscribe命令,功能上很相似,但存储和处理上是不同的;如果不熟悉很容易造成数据的混乱以及理解上的偏差.

1. 数据结构

subscribe命令普通订阅模式: 使用hash结构存储

server.pubsub_channels

psubscribe命令正则订阅模式: 使用list结构存储

server.pubsub_patterns

2. 信息发布处理

在处理发布消息时,也是两种模式分别处理发送的.

int pubsubpublishmessage(robj *channel, robj *message) { ... /* send to clients listening for that channel */ de = dictfind(server.pubsub_channels,channel); if (de) { ... } /* send to clients listening to matching channels */ if (listlength(server.pubsub_patterns)) { ... } return receivers;}

3. pubsub 查询命令

在pubsub 查询子命令也是区分了不同通道的.

void pubsubcommand(client *c) { if (!strcasecmp(c->argv[1]->ptr,"channels") && (c->argc == 2 || c->argc ==3)) { ... dictiterator *di = dictgetiterator(server.pubsub_channels); ... } else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) { /* pubsub numsub [channel_1 ... channel_n] */ ... list *l = dictfetchvalue(server.pubsub_channels,c->argv[j]); ... } else if (!strcasecmp(c->argv[1]->ptr,"numpat") && c->argc == 2) { /* pubsub numpat */ addreplylonglong(c,listlength(server.pubsub_patterns)); } else { ... }}

pub/sub对系统解耦有很大帮助,但也有些不足:

1. 订阅客户端消息的消费速度却不够快的话,那么不断积压的消息会使redis输出缓冲区的体积变得越来越大,这可能使得redis本身的速度变慢,甚至直接崩溃.

2.如果订阅客户端断线,那么他将会丢失所有断线期间发布的信息.这个问题在redis v5.0版本中会有stream相关api代替.

推荐阅读

---

• 
  

• 
  

(长按二维码即可关注)