Redis 消息订阅（MessageListener接口）

た入场券 2024-03-16 11:51 26阅读 0赞

在消息接收端或消息消费端，Spring Data Redis 可以通过直接命名或使用模式匹配订阅一个或多个频道（Channel）。

模式匹配方式非常有用，因为它不仅允许使用一个命令创建多个订阅，还可以侦听订阅时尚未创建的频道（只要它们与模式匹配）。例如：

（1）订阅消息，模式为 \*.hxstrive.com，可以匹配任何以 .hxstrive.com 结尾的模式，如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 </td> 
 <td> <code>D:\server\redis-x64-5.0.14.1&gt;&nbsp;redis-cli</code> <code>127.0.0.1:6379&gt;&nbsp;psubscribe&nbsp;*.hxstrive.com</code> <code>Reading&nbsp;messages...&nbsp;(press&nbsp;Ctrl-C&nbsp;to&nbsp;quit)</code> <code>1)&nbsp;</code><code>"psubscribe"</code> <code>2)&nbsp;</code><code>"*.hxstrive.com"</code> <code>3)&nbsp;(integer)&nbsp;1</code> <code>1)&nbsp;</code><code>"pmessage"</code> <code>2)&nbsp;</code><code>"*.hxstrive.com"</code> <code>3)&nbsp;</code><code>"www.hxstrive.com"</code> <code>4)&nbsp;</code><code>"hello,&nbsp;www.hxstrive.com"</code> <code>1)&nbsp;</code><code>"pmessage"</code> <code>2)&nbsp;</code><code>"*.hxstrive.com"</code> <code>3)&nbsp;</code><code>"doc.hxstrive.com"</code> <code>4)&nbsp;</code><code>"hello,&nbsp;doc.hxstrive.com"</code> </td> 
 </tr> 
 </tbody> 
</table>

（2）发送消息，分别给 www.hxstrive.com 和 doc.hxstrive.com 模式，如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 </td> 
 <td> <code>D:\server\redis-x64-5.0.14.1&gt;&nbsp;redis-cli</code> <code>127.0.0.1:6379&gt;&nbsp;publish&nbsp;www.hxstrive.com&nbsp;</code><code>"hello,&nbsp;www.hxstrive.com"</code> <code>(integer)&nbsp;1</code> <code>127.0.0.1:6379&gt;&nbsp;publish&nbsp;doc.hxstrive.com&nbsp;</code><code>"hello,&nbsp;doc.hxstrive.com"</code> <code>(integer)&nbsp;1</code> <code>127.0.0.1:6379&gt;</code> </td> 
 </tr> 
 </tbody> 
</table>

在底层，RedisConnection 提供了 subscribe 和 pSubscribe 方法，分别映射 Redis 按频道订阅 subscribe 命令，按模式订阅 psubscribe 命令。它们语法如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 </td> 
 <td> <code>subscribe&nbsp;channel&nbsp;[channel&nbsp;...]</code> <code>psubscribe&nbsp;pattern&nbsp;[pattern&nbsp;...]</code> </td> 
 </tr> 
 </tbody> 
</table>

其中：

*  channel 表示频道名称
 *  pattern 表示模式字符串

请注意，可以将多个频道（Channel）或多个匹配模式用作参数。为了更改连接的订阅或查询它是否正在侦听，RedisConnection 提供了getSubscription() 和 isSubscribed() 方法。

Spring Data Redis 中的订阅命令是阻塞模式。也就是说，在连接上调用 subscribe 会导致当前线程阻塞，等待消息。只有当订阅被取消时，线程才会被释放，你可以在另一个线程上，对订阅消息的连接调用 unsubscribe 或 pUnsubscrribe 方法释放。

如前所述，一旦订阅，连接就会开始等待消息。此时，连接仅允许添加新的订阅、修改现有订阅、取消现有订阅的命令。如果调用 subscribe、pSubscribe、unsubscrribe 或 pUnsubscrube 之外的任何命令都会引发异常。

为了订阅消息，需要实现 MessageListener 回调。每次新消息到达时，都会调用回调，并通过 onMessage 方法执行用户业务代码。该接口不仅可以访问实际消息，还可以访问接收消息的频道（Channel），以及订阅时用于匹配频道（Channel）的模式。此信息使被调用者不仅可以通过内容区分各种消息，还可以检查其他详细信息。

MessageListener 接口源码如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 </td> 
 <td> <code>package</code>&nbsp;<code>org.springframework.data.redis.connection;</code> <code>import</code>&nbsp;<code>org.springframework.lang.Nullable;</code> <code>/**</code> <code>&nbsp;</code><code>*&nbsp;Listener&nbsp;of&nbsp;messages&nbsp;published&nbsp;in&nbsp;Redis.</code> <code>&nbsp;</code><code>*&nbsp;在&nbsp;Redis&nbsp;上发布的消息的监听者。</code> <code>&nbsp;</code><code>*&nbsp;@author&nbsp;Costin&nbsp;Leau</code> <code>&nbsp;</code><code>*&nbsp;@author&nbsp;Christoph&nbsp;Strobl</code> <code>&nbsp;</code><code>*/</code> <code>public</code>&nbsp;<code>interface</code>&nbsp;<code>MessageListener&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;Callback&nbsp;for&nbsp;processing&nbsp;received&nbsp;objects&nbsp;through&nbsp;Redis.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;通过&nbsp;Redis&nbsp;处理接收到的对象的回调。</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@param&nbsp;message&nbsp;message&nbsp;must&nbsp;not&nbsp;be&nbsp;{@literal&nbsp;null}.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;消息对象，不能为&nbsp;null</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@param&nbsp;pattern&nbsp;pattern&nbsp;matching&nbsp;the&nbsp;channel&nbsp;(if&nbsp;specified)&nbsp;-&nbsp;can&nbsp;be&nbsp;{@literal&nbsp;null}.</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;与通道匹配的模式(如果指定)-可以为空。</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>void</code>&nbsp;<code>onMessage(Message&nbsp;message,&nbsp;</code><code>@Nullable</code>&nbsp;<code>byte</code><code>[]&nbsp;pattern);</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

### 消息侦听器容器 ###

由于其阻塞特性，低级订阅不具吸引力，因为它需要对每个侦听器进行连接和线程管理。为了缓解这个问题，Spring Data 提供了RedisMessageListenerContainer 类，它可以完成所有繁重的工作。如果您熟悉 EJB 和 JMS，您应该会发现熟悉的概念，因为它被设计为尽可能接近 Spring Framework 消息驱动 POJO（MDP，message-driven POJOs）中的支持。

RedisMessageListenerContainer 充当消息侦听器容器。它用于接收来自 Redis 频道（Channel）的消息，并驱动注入其中的MessageListener 实例。侦听器容器负责消息接收的所有线程，并将消息发送到侦听器进行处理。消息侦听器容器是 MDP 和消息传递提供者之间的中介，负责注册以接收消息、资源获取和释放、异常转换等。这让您作为应用程序开发人员可以编写与接收消息相关的业务逻辑，并将样板 Redis 基础设施关注点委托给框架。

MessageListener 还可以实现 SubscriptionListener，以便在确认订阅/取消订阅时接收通知。同步调用时，侦听订阅通知非常有用。

此外，为了最小化应用程序占用空间，RedisMessageListenerContainer 允许多个侦听器共享一个连接和一个线程，即使它们不共享频道（Channel）。因此，无论应用程序跟踪多少侦听器或频道（Channel），运行时成本在其整个生命周期中都保持不变。此外，容器允许运行时更改配置，以便您可以在应用程序运行时添加或删除侦听器，而无需重新启动。此外，容器使用延迟订阅方法，仅在需要时使用RedisConnection。如果所有侦听器都被取消订阅，则会自动执行清理，并释放线程。

为了帮助处理消息的异步特性，容器需要一个java.util.concurrent.Executor（或 Spring 的 TaskExecutor）来分派消息。根据负载、侦听器数量或运行时环境的不同，您应该更改或调整执行器以更好地满足您的需求。特别是，在托管环境（如应用程序服务器）中，强烈建议选择合适的 TaskExecutor 来利用其运行。

### MessageListenerAdapter ###

MessageListenerAdapter 类是 Spring 异步消息传递支持中的最后一个组件。简而言之，它允许您将几乎任何类公开为 MDP（尽管存在一些约束）。

### 示例 ###

使用 Spring Data Redis 订阅“hxstrive”频道（称为“客户端”），然后利用 redis-cli 工具的 publish 命令向“hxstrive”频道推送一条消息，Redis 将频道收到的消息转发给客户端，最后客户端将消息内容打印到控制台。

#### 配置 RedisTemplate ####

#### 开发订阅服务 ####

自定义 MessageListener 接口实现，来开发一个订阅服务。代码如下：

#### 消息配置 ####

配置 RedisMessageListenerContainer 监听器容器、消息侦听器适配器和频道。代码如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 </td> 
 <td> <code>import</code>&nbsp;<code>org.springframework.beans.factory.annotation.Autowired;</code> <code>import</code>&nbsp;<code>org.springframework.context.annotation.Bean;</code> <code>import</code>&nbsp;<code>org.springframework.context.annotation.Configuration;</code> <code>import</code>&nbsp;<code>org.springframework.data.redis.connection.RedisConnectionFactory;</code> <code>import</code>&nbsp;<code>org.springframework.data.redis.listener.PatternTopic;</code> <code>import</code>&nbsp;<code>org.springframework.data.redis.listener.RedisMessageListenerContainer;</code> <code>import</code>&nbsp;<code>org.springframework.data.redis.listener.Topic;</code> <code>import</code>&nbsp;<code>org.springframework.data.redis.listener.adapter.MessageListenerAdapter;</code> <code>import</code>&nbsp;<code>java.util.ArrayList;</code> <code>import</code>&nbsp;<code>java.util.List;</code> <code>/**</code> <code>&nbsp;</code><code>*&nbsp;消息配置</code> <code>&nbsp;</code><code>*&nbsp;@author&nbsp;hxstrive.com&nbsp;2022/10/13</code> <code>&nbsp;</code><code>*/</code> <code>@Configuration</code> <code>public</code>&nbsp;<code>class</code>&nbsp;<code>MessageConfig&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Autowired</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>Subscriber&nbsp;subscriber;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;配置&nbsp;RedisMessageListenerContainer&nbsp;监听器容器</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@param&nbsp;connectionFactory&nbsp;连接工厂</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@param&nbsp;listenerAdapter&nbsp;消息侦听器适配器</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@return</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Bean</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>RedisMessageListenerContainer&nbsp;container(RedisConnectionFactory&nbsp;connectionFactory,</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>MessageListenerAdapter&nbsp;listenerAdapter)&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>RedisMessageListenerContainer&nbsp;container&nbsp;=&nbsp;</code><code>new</code>&nbsp;<code>RedisMessageListenerContainer();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>container.setConnectionFactory(connectionFactory);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//&nbsp;消息主题列表</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>List&lt;Topic&gt;&nbsp;topicList&nbsp;=&nbsp;</code><code>new</code>&nbsp;<code>ArrayList&lt;&gt;();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//&nbsp;订阅&nbsp;“hxstrive”&nbsp;频道</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>topicList.add(</code><code>new</code>&nbsp;<code>ChannelTopic(</code><code>"hxstrive"</code><code>));</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//&nbsp;订阅&nbsp;“*.hxstrive.com”&nbsp;模式</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>topicList.add(</code><code>new</code>&nbsp;<code>PatternTopic(</code><code>"*.hxstrive.com"</code><code>));</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>container.addMessageListener(listenerAdapter,&nbsp;topicList);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>container;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>//&nbsp;将消息侦听器添加到容器中</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>container.addMessageListener(listenerAdapter,&nbsp;topicList);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>container;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>/**</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;消息侦听器适配器，能将消息委托给目标侦听器</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*&nbsp;@return</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>*/</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Bean</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>MessageListenerAdapter&nbsp;listenerAdapter()&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>new</code>&nbsp;<code>MessageListenerAdapter(subscriber);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

#### 启动类 ####

启动类代码如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 </td> 
 <td> <code>import</code>&nbsp;<code>org.springframework.boot.SpringApplication;</code> <code>import</code>&nbsp;<code>org.springframework.boot.autoconfigure.SpringBootApplication;</code> <code>@SpringBootApplication</code> <code>public</code>&nbsp;<code>class</code>&nbsp;<code>RedisLearnSpringDataApplication&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>static</code>&nbsp;<code>void</code>&nbsp;<code>main(String[]&nbsp;args)&nbsp;{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>SpringApplication.run(RedisLearnSpringDataApplication.</code><code>class</code><code>,&nbsp;args);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

启动 Spring Boot 服务，如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 </td> 
 <td> <code>&nbsp;&nbsp;</code><code>.&nbsp;&nbsp;&nbsp;____&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;_&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;__&nbsp;_&nbsp;_</code> <code>&nbsp;</code><code>/\\&nbsp;/&nbsp;___'_&nbsp;__&nbsp;_&nbsp;_(_)_&nbsp;__&nbsp;&nbsp;__&nbsp;_&nbsp;\&nbsp;\&nbsp;\&nbsp;\</code> <code>(&nbsp;(&nbsp;)\___&nbsp;|&nbsp;'_&nbsp;|&nbsp;'_|&nbsp;|&nbsp;'_&nbsp;\/&nbsp;_`&nbsp;|&nbsp;\&nbsp;\&nbsp;\&nbsp;\</code> <code>&nbsp;</code><code>\\/&nbsp;&nbsp;___)|&nbsp;|_)|&nbsp;|&nbsp;|&nbsp;|&nbsp;|&nbsp;||&nbsp;(_|&nbsp;|&nbsp;&nbsp;)&nbsp;)&nbsp;)&nbsp;)</code> <code>&nbsp;&nbsp;</code><code>'&nbsp;&nbsp;|____|&nbsp;.__|_|&nbsp;|_|_|&nbsp;|_\__,&nbsp;|&nbsp;/&nbsp;/&nbsp;/&nbsp;/</code> <code>&nbsp;</code><code>=========|_|==============|___/=/_/_/_/</code> <code>&nbsp;</code><code>::&nbsp;Spring&nbsp;Boot&nbsp;::&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;(v2.1.7.RELEASE)</code> <code>...</code> <code>2022-10-17&nbsp;12:38:34.292&nbsp;&nbsp;INFO&nbsp;2932&nbsp;---&nbsp;[&nbsp;&nbsp;restartedMain]&nbsp;o.s.b.w.embedded.tomcat.TomcatWebServer&nbsp;&nbsp;:&nbsp;Tomcat&nbsp;started&nbsp;on&nbsp;port(s):&nbsp;8080&nbsp;(http)&nbsp;with&nbsp;context&nbsp;path&nbsp;''</code> <code>2022-10-17&nbsp;12:38:34.292&nbsp;&nbsp;INFO&nbsp;2932&nbsp;---&nbsp;[&nbsp;&nbsp;restartedMain]&nbsp;c.h.r.RedisLearnSpringDataApplication&nbsp;&nbsp;&nbsp;&nbsp;:&nbsp;Started&nbsp;RedisLearnSpringDataApplication&nbsp;in&nbsp;6.267&nbsp;seconds&nbsp;(JVM&nbsp;running&nbsp;for&nbsp;10.329)</code> </td> 
 </tr> 
 </tbody> 
</table>

#### 发送消息 ####

使用 redis-cli 命令连接到 Redis，然后使用 publish 命令向“hxstrive”频道发送一条消息。如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 </td> 
 <td> <code>D:\server\redis-x64-5.0.14.1&gt;redis-cli</code> <code>127.0.0.1:6379&gt;&nbsp;publish&nbsp;hxstrive&nbsp;</code><code>"hi!&nbsp;spring&nbsp;data&nbsp;redis"</code> <code>(integer)&nbsp;1</code> <code>127.0.0.1:6379&gt;&nbsp;publish&nbsp;doc.hxstrive.com&nbsp;</code><code>"hi!&nbsp;spring&nbsp;data&nbsp;redis"</code> <code>(integer)&nbsp;1</code> <code>127.0.0.1:6379&gt;&nbsp;publish&nbsp;www.hxstrive.com&nbsp;</code><code>"hi!&nbsp;spring&nbsp;data&nbsp;redis"</code> <code>(integer)&nbsp;1</code> <code>127.0.0.1:6379&gt;</code> </td> 
 </tr> 
 </tbody> 
</table>

此时，客户端将收到如下消息：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 </td> 
 <td> <code>频道:&nbsp;hxstrive</code> <code>消息内容:&nbsp;hi!&nbsp;spring&nbsp;data&nbsp;redis</code> <code>频道:&nbsp;doc.hxstrive.com</code> <code>消息内容:&nbsp;hi!&nbsp;spring&nbsp;data&nbsp;redis</code> <code>频道:&nbsp;www.hxstrive.com</code> <code>消息内容:&nbsp;hi!&nbsp;spring&nbsp;data&nbsp;redis</code> </td> 
 </tr> 
 </tbody> 
</table>