Pub/Sub 消息传递

Spring Data 为 Redis 提供了专用的消息传递集成，其功能和命名与 Spring Framework 中的 JMS 集成类似。spring-doc.cn

Redis 消息传递大致可分为两个功能领域：spring-doc.cn

这是通常称为 Publish/Subscribe（简称 Pub/Sub）的模式的一个示例。该类用于消息生成。对于类似于 Java EE 的消息驱动 Bean 样式的异步接收，Spring Data 提供了一个专用的消息侦听器容器，用于创建消息驱动的 POJO （MDP），对于同步接收，则提供 Contract。RedisTemplateRedisConnectionspring-doc.cn

和 软件包为 Redis 消息传递提供核心功能。org.springframework.data.redis.connectionorg.springframework.data.redis.listenerspring-doc.cn

发布 （发送消息）

要发布消息，与其他操作一样，您可以使用 low-level 或 high-level 。这两个实体都提供方法，该方法接受消息和目标通道作为参数。虽然需要原始数据（字节数组），但允许将任意对象作为消息传入，如以下示例所示：[Reactive]RedisConnection[Reactive]RedisOperationspublishRedisConnection[Reactive]RedisOperationsspring-doc.cn

// send message through connection
RedisConnection con = …
byte[] msg = …
byte[] channel = …
con.pubSubCommands().publish(msg, channel);

// send message through RedisOperations
RedisOperations operations = …
Long numberOfClients = operations.convertAndSend("hello!", "world");
// send message through connection
ReactiveRedisConnection con = …
ByteBuffer[] msg = …
ByteBuffer[] channel = …
con.pubSubCommands().publish(msg, channel);

// send message through ReactiveRedisOperations
ReactiveRedisOperations operations = …
Mono<Long> numberOfClients = operations.convertAndSend("hello!", "world");

订阅 （接收消息）

在接收方，可以通过直接命名或使用模式匹配来订阅一个或多个通道。后一种方法非常有用，因为它不仅允许使用一个命令创建多个订阅，还可以侦听在订阅时尚未创建的频道（只要它们与模式匹配）。spring-doc.cn

在低级别，提供 和 方法，分别映射用于按通道或模式订阅的 Redis 命令。请注意，多个通道或 pattern 可以用作参数。要更改连接的订阅或查询它是否正在侦听，请提供 and 方法。RedisConnectionsubscribepSubscribeRedisConnectiongetSubscriptionisSubscribedspring-doc.cn

Spring Data Redis 中的订阅命令被阻塞。也就是说，在连接上调用 subscribe 会导致当前线程在开始等待消息时阻塞。仅当取消订阅时，才会释放线程，当另一个线程调用或在同一连接上时，会发生这种情况。有关此问题的解决方案，请参阅“消息侦听器容器”（本文档后面的部分）。unsubscribepUnsubscribe

如前所述，订阅后，连接将开始等待消息。仅允许使用添加新订阅、修改现有订阅和取消现有订阅的命令。调用 、、 、 或 之外的任何内容都会引发异常。subscribepSubscribeunsubscribepUnsubscribespring-doc.cn

为了订阅消息，需要实现回调。每次收到新消息时，都会调用回调，并且该方法会运行用户代码。该接口不仅可以访问实际消息，还可以访问接收消息的通道以及订阅用于匹配通道的模式（如果有）。此信息使被调用方不仅可以通过内容来区分各种消息，还可以检查其他详细信息。MessageListeneronMessagespring-doc.cn

消息侦听器容器

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

RedisMessageListenerContainer 充当消息侦听器容器。它用于从 Redis 通道接收消息，并驱动注入其中的 MessageListener 实例。侦听器容器负责消息接收的所有线程处理，并分派到侦听器中进行处理。消息侦听器容器是 MDP 和消息传递提供商之间的中介，负责注册以接收消息、资源获取和释放、异常转换等。这样，作为应用程序开发人员，您可以编写与接收消息（并对其做出反应）相关的（可能复杂的）业务逻辑，并将样板 Redis 基础设施问题委托给框架。spring-doc.cn

MessageListener 还可以实现 SubscriptionListener 以在订阅/取消订阅确认时接收通知。在同步调用时，侦听订阅通知可能很有用。spring-doc.cn

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

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

The MessageListenerAdapter

MessageListenerAdapter类是 Spring 异步消息传递支持中的最后一个组件。简而言之，它允许您将几乎任何类公开为 MDP（尽管有一些限制）。spring-doc.cn

请考虑以下接口定义：spring-doc.cn

public interface MessageDelegate {
  void handleMessage(String message);
  void handleMessage(Map message);
  void handleMessage(byte[] message);
  void handleMessage(Serializable message);
  // pass the channel/pattern as well
  void handleMessage(Serializable message, String channel);
 }

请注意，尽管接口不扩展接口，但仍可以通过使用 MessageListenerAdapter 类将其用作 MDP。还要注意各种消息处理方法是如何根据它们可以接收和处理的各种类型的内容进行强类型化的。此外，将消息发送到的通道或模式可以作为 type 的第二个参数传递给方法：MessageListenerMessageStringspring-doc.cn

public class DefaultMessageDelegate implements MessageDelegate {
  // implementation elided for clarity...
}
Notice how the above implementation of the `MessageDelegate` interface (the above `DefaultMessageDelegate` class) has *no* Redis dependencies at all. It truly is a POJO that we make into an MDP with the following configuration:
@Configuration
class MyConfig {

  // …

  @Bean
  DefaultMessageDelegate listener() {
    return new DefaultMessageDelegate();
  }

  @Bean
  MessageListenerAdapter messageListenerAdapter(DefaultMessageDelegate listener) {
    return new MessageListenerAdapter(listener, "handleMessage");
  }

  @Bean
  RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory, MessageListenerAdapter listener) {

    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.addMessageListener(listener, ChannelTopic.of("chatroom"));
    return container;
  }
}
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xmlns:redis="http://www.springframework.org/schema/redis"
   xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
   http://www.springframework.org/schema/redis https://www.springframework.org/schema/redis/spring-redis.xsd">

<!-- the default ConnectionFactory -->
<redis:listener-container>
  <!-- the method attribute can be skipped as the default method name is "handleMessage" -->
  <redis:listener ref="listener" method="handleMessage" topic="chatroom" />
</redis:listener-container>

<bean id="listener" class="redisexample.DefaultMessageDelegate"/>
 ...
</beans>
侦听器主题可以是通道（例如，）或模式（例如，topic="chatroom"topic="*room")

前面的示例使用 Redis 命名空间声明消息侦听器容器，并自动将 POJO 注册为侦听器。完整的 bean 定义如下：spring-doc.cn

<bean id="messageListener" class="org.springframework.data.redis.listener.adapter.MessageListenerAdapter">
  <constructor-arg>
    <bean class="redisexample.DefaultMessageDelegate"/>
  </constructor-arg>
</bean>

<bean id="redisContainer" class="org.springframework.data.redis.listener.RedisMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory"/>
  <property name="messageListeners">
    <map>
      <entry key-ref="messageListener">
        <bean class="org.springframework.data.redis.listener.ChannelTopic">
          <constructor-arg value="chatroom"/>
        </bean>
      </entry>
    </map>
  </property>
</bean>

每次收到消息时，适配器都会自动透明地在低级格式和所需对象类型之间执行转换（使用 configured）。由方法调用导致的任何异常都由容器捕获和处理（默认情况下，会记录异常）。RedisSerializerspring-doc.cn

反应式消息侦听器容器

Spring Data 提供了 ReactiveRedisMessageListenerContainer，它代表用户完成了转换和订阅状态管理的所有繁重工作。spring-doc.cn

消息侦听器容器本身不需要外部线程资源。它使用驱动程序线程来发布消息。spring-doc.cn

ReactiveRedisConnectionFactory factory = …
ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(factory);

Flux<ChannelMessage<String, String>> stream = container.receive(ChannelTopic.of("my-channel"));

要等待并确保正确订阅，您可以使用返回 . 由于完成了对给定主题的订阅，结果以内部发布者完成。通过拦截信号，您可以同步服务器端订阅。receiveLaterMono<Flux<ChannelMessage>>MonoonNextspring-doc.cn

ReactiveRedisConnectionFactory factory = …
ReactiveRedisMessageListenerContainer container = new ReactiveRedisMessageListenerContainer(factory);

Mono<Flux<ChannelMessage<String, String>>> stream = container.receiveLater(ChannelTopic.of("my-channel"));

stream.doOnNext(inner -> // notification hook when Redis subscriptions are synchronized with the server)
    .flatMapMany(Function.identity())
    .…;

通过模板 API 订阅

如上所述，您可以直接使用 ReactiveRedisTemplate 订阅频道/模式。这种方法 提供简单但有限的解决方案，因为您失去了在初始 的。尽管如此，你仍然可以通过返回的 using eg..什么时候 读取完毕，出错或取消时，将再次释放所有绑定的资源。Fluxtake(Duration)spring-doc.cn

redisTemplate.listenToChannel("channel1", "channel2").doOnNext(msg -> {
    // message processing ...
}).subscribe();