redis pub/sub原理及实战

客官°小女子只卖身不卖艺 2023-01-14 01:48 108阅读 0赞

**目录**

概念

原理

发布

使用

Subscribing

Reactive API

Redis Cluster

--------------------

## 概念 ##

Redis同样支持消息的发布/订阅（Pub/Sub）模式，这和中间件activemq有些类似。订阅者（Subscriber）可以订阅自己感兴趣的频道（Channel），发布者（Publisher）可以将消息发往指定的频道（Channel），正式通过这种方式，可以将消息的发送者和接收者解耦。另外，由于可以动态的Subscribe和Unsubscribe，也可以提高系统的灵活性和可扩展性。本次我们介绍lettuce的实现

## 原理 ##

client->pubsub\_channels 是客户端维护的一个以dict结构的维护的订阅频道哈希表，VAL是NULL,不需要值。

server->pubsub\_channels 是服务端维护的一个以dict结构的维护的订阅频道哈希表，VAL是以client维护的双向链表adlist。

### 订阅 ###

订阅流程：SUBSCRIBE命令   SUBSCRIBE channel \[channel ...\]

1.首先是将当前订阅的频道channel添加进客户端的pubsub\_channels哈希表里面。

2.然后在将当前订阅的频道channel和对应的client以键值对添加服务端的pubsub\_channels的哈希表里。

3.最后将返回的信息返回给客户端。

### 发布 ###

发布流程：PUBLISH命令   PUBLISH channel message

1.首先通过频道在服务端的pubsub\_channels哈希表里面找到对应的客户端链表。

2.然后递归循环链表，逐个将消息message发送对应订阅的客户端。

3.正则匹配的频道  逐个发送对应订阅的客户端。

## 使用 ##

### Subscribing ###

一个连接可以通知到多个实现了RedisPubSubListener（Lettuce提供了更方便的RedisPubSubAdapter）的linstener。所有的listener都关联在`StatefulRedisPubSubConnection`/`StatefulRedisClusterConnection`.

Example 1. Synchronous subscription

<table> 
 <tbody> 
  <tr> 
   <td> <p><code>StatefulRedisPubSubConnection&lt;String, String&gt; connection = client.connectPubSub()</code></p> <p><code>connection.addListener(</code><code>new</code>&nbsp;<code>RedisPubSubListener&lt;String, String&gt;() { ... })</code></p> <p>&nbsp;</p> <p><code>RedisPubSubCommands&lt;String, String&gt; sync = connection.sync();</code></p> <p><code>sync.subscribe(</code><code>"channel"</code><code>);</code></p> </td> 
  </tr> 
 </tbody> 
</table>

Example 2. Asynchronous subscription

<table> 
 <tbody> 
  <tr> 
   <td> <p><code>StatefulRedisPubSubConnection&lt;String, String&gt; connection = client.connectPubSub()</code></p> <p><code>connection.addListener(</code><code>new</code>&nbsp;<code>RedisPubSubListener&lt;String, String&gt;() { ... })</code></p> <p>&nbsp;</p> <p><code>RedisPubSubAsyncCommands&lt;String, String&gt; async = connection.async();</code></p> <p><code>RedisFuture&lt;Void&gt; future = async.subscribe(</code><code>"channel"</code><code>);</code></p> </td> 
  </tr> 
 </tbody> 
</table>

### Reactive API ###

响应式API提供了hot `Observable`s来监听`ChannelMessage`s和`PatternMessage`s。`Observable`s接受所有入站消息。你可以使用observable chain 进行消息过滤。Observable会停止触发世界当订阅者取消订阅。

Example 3. Reactive subscription

<table> 
 <tbody> 
  <tr> 
   <td> <p><code>StatefulRedisPubSubConnection&lt;String, String&gt; connection = client.connectPubSub()</code></p> <p>&nbsp;</p> <p><code>RedisPubSubReactiveCommands&lt;String, String&gt; reactive = connection.reactive();</code></p> <p><code>reactive.subscribe(</code><code>"channel"</code><code>).subscribe();</code></p> <p>&nbsp;</p> <p><code>reactive.observeChannels().doOnNext(patternMessage -&gt; {...}).subscribe()</code></p> </td> 
  </tr> 
 </tbody> 
</table>

### Redis Cluster ###

Redis 集群支持Publish/Subscribe 但是需要注意一些事项。User-space的发布/订阅消息（称为PUBLISH）在整个集群中广播，而与订阅特定频道/模式无关。发布行为会连接到集群中的所有节点，而不用客户端连接到所有节点都会收到订阅消息。

RedisClusterClient.connectPubSub（）提供了一个支持群集的Pub / Sub连接，该连接允许监听集群的重新配置并在拓扑更改时重新连接。

Example 4. Redis Cluster Publish/Subscribe

<table> 
 <tbody> 
  <tr> 
   <td> <p><code>StatefulRedisClusterPubSubConnection&lt;String, String&gt; connection = clusterClient.connectPubSub()</code></p> <p><code>connection.addListener(</code><code>new</code>&nbsp;<code>RedisPubSubListener&lt;String, String&gt;() { ... })</code></p> <p>&nbsp;</p> <p><code>RedisPubSubCommands&lt;String, String&gt; sync = connection.sync();</code></p> <p><code>sync.subscribe(</code><code>"channel"</code><code>);</code></p> </td> 
  </tr> 
 </tbody> 
</table>

Redis Cluster also makes a distinction between user-space and key-space messages. Key-space notifications (Pub/Sub messages for key-activity) stay node-local and are not broadcasted across the Redis Cluster. A notification about, e.g. an expiring key, stays local to the node on which the key expired.

Clients that are interested in keyspace notifications must subscribe to the appropriate node (or nodes) to receive these notifications. You can either use `RedisClient.connectPubSub()` to establish Pub/Sub connections to the individual nodes or use `RedisClusterClient`'s message propagation and NodeSelection API to get a managed set of connections.

Example 5. Redis Cluster Publish/Subscribe with node message propagation

<table> 
 <tbody> 
  <tr> 
   <td> <p><code>StatefulRedisClusterPubSubConnection&lt;String, String&gt; connection = clusterClient.connectPubSub()</code></p> <p><code>connection.addListener(</code><code>new</code>&nbsp;<code>RedisClusterPubSubListener&lt;String, String&gt;() { ... })</code></p> <p><code>connection.setNodeMessagePropagation(</code><code>true</code><code>);</code></p> <p>&nbsp;</p> <p><code>RedisPubSubCommands&lt;String, String&gt; sync = connection.sync();</code></p> <p><code>sync.masters().commands().subscribe(</code><code>"__keyspace@0__:*"</code><code>);</code></p> </td> 
  </tr> 
 </tbody> 
</table>

There are two things to pay special attention to:

1.  Replication: Keys replicated to replica nodes, especially considering expiry, generate keyspace events on all nodes holding the key. If a key expires and it is replicated, it will expire on the master and all replicas. Each Redis server will emit keyspace events. Subscribing to non-master nodes, therefore, will let your application see multiple events of the same type for the same key because of Redis distributed nature.
2.  Topology Changes: Subscriptions are issued either by using the NodeSelection API or by calling `subscribe(…)` on the individual cluster node connections. Subscription registrations are not propagated to new nodes that are added on a topology change.