做简单的网站首页,有没有教做帽子的网站,免费网站建站模块,乐清公司做网站【Spring连载】使用Spring Data访问Redis#xff08;八#xff09;----发布/订阅消息Pub/Sub Messaging 一、发布消息Publishing (Sending Messages)二、订阅消息Subscribing (Receiving Messages)2.1 消息监听容器Message Listener Containers2.2 消息监听适配器The Message… 【Spring连载】使用Spring Data访问Redis八----发布/订阅消息Pub/Sub Messaging 一、发布消息Publishing (Sending Messages)二、订阅消息Subscribing (Receiving Messages)2.1 消息监听容器Message Listener Containers2.2 消息监听适配器The MessageListenerAdapter 三、反应式消息监听器容器Reactive Message Listener Container3.1 通过template API订阅Subscribing via template API Spring Data为Redis提供了专用的消息集成在功能和命名方面与Spring Framework中的JMS集成类似。 Redis消息传递大致可以分为两个功能领域 消息的发布publish或生产消息的订阅subscribe或消费
这是一个通常称为发布/订阅简称Pub/Sub的模式示例。RedisTemplate类用于消息生成。对于类似于Java EE的消息驱动bean风格的异步接收Spring Data提供了一个专用的消息监听器容器用于创建消息驱动的POJOMDP对于同步接收还提供RedisConnection。 org.springframework.data.redis.connection 和 org.springframework.data.redis.listener包提供Redis消息传递的核心功能。
一、发布消息Publishing (Sending Messages)
要发布消息你可以像使用其他操作一样使用低级的RedisConnection或高级的RedisOperations。这两个实体都提供了publish方法该方法接受消息和目标通道channel作为参数。RedisConnection需要原始数据(字节数组)而RedisOperations允许任意对象作为消息传递如下面的例子所示:
// 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);一个相对完整的发布例子
package com.example.demo;import org.springframework.data.redis.connection.RedisPassword;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;public class RedisPublishMessage {public JedisConnectionFactory jedisConnectionFactory() {RedisStandaloneConfiguration redisStandaloneConfiguration new RedisStandaloneConfiguration();redisStandaloneConfiguration.setHostName(localhost);redisStandaloneConfiguration.setDatabase(0);redisStandaloneConfiguration.setPassword(RedisPassword.of(123456));redisStandaloneConfiguration.setPort(6379);return new JedisConnectionFactory(redisStandaloneConfiguration);}public static void main(String[] args) {JedisConnectionFactory connectionFactory new RedisApplication().jedisConnectionFactory();connectionFactory.afterPropertiesSet();RedisTemplateString, String template new RedisTemplate();template.setConnectionFactory(connectionFactory);template.setDefaultSerializer(StringRedisSerializer.UTF_8);template.afterPropertiesSet();
// send message through RedisOperationsRedisOperations operations template;Long numberOfClients operations.convertAndSend(mychannel, This is Gabriel);System.out.println(numberOfClients);}
}二、订阅消息Subscribing (Receiving Messages)
在接收端可以通过直接命名或使用模式(pattern)匹配来订阅一个或多个通道(channels)。后一种方法非常有用因为它不仅允许使用一个命令创建多个订阅subscription而且还可以监听订阅时尚未创建的通道只要它们与模式匹配。 在底层RedisConnection提供了subscribe和pSubscribe方法它们分别映射Redis命令的以通道或模式进行订阅。前述2个方法可以使用多个通道或多个模式做为参数。为了更改连接的订阅或检查连接是否在监听RedisConnection提供了getSubscription和isSubscribed方法。 Spring Data Redis中的Subscription命令是阻塞的。也就是说在连接上调用subscribe会导致当前线程在开始等待消息时阻塞。只有当订阅被取消时线程才会被释放当另一个线程在同一连接上调用unsubscribe或pUnsubscribe时就会发生这种情况。有关此问题的解决方案请参阅“2.1 消息监听容器”本文档稍后部分。 如前所述一旦订阅连接就会开始等待消息。只允许添加新订阅、修改现有订阅和取消现有订阅的命令。调用除subscribe, pSubscribe, unsubscribe, 和 pUnsubscribe之外的任何操作都会引发异常。 为了订阅消息需要实现MessageListener回调。每次新消息到达时都会调用回调并通过onMessage方法运行用户代码。该接口不仅可以访问实际消息还可以访问接收该消息的通道以及用于匹配订阅的通道的模式如果有的话。这些信息使被调用者来区分各种消息不仅是通过内容还可以通过检查其他细节。 一个相对完整的订阅例子
package com.example.demo;import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.jedis.JedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;import java.nio.charset.StandardCharsets;public class RedisSubscribeMessage {public JedisConnectionFactory jedisConnectionFactory() {RedisStandaloneConfiguration redisStandaloneConfiguration new RedisStandaloneConfiguration();redisStandaloneConfiguration.setHostName(localhost);redisStandaloneConfiguration.setDatabase(0);redisStandaloneConfiguration.setPassword(RedisPassword.of(123456));redisStandaloneConfiguration.setPort(6379);return new JedisConnectionFactory(redisStandaloneConfiguration);}public static void main(String[] args) {JedisConnectionFactory connectionFactory new RedisApplication().jedisConnectionFactory();connectionFactory.afterPropertiesSet();RedisTemplateString, String template new RedisTemplate();template.setConnectionFactory(connectionFactory);template.setDefaultSerializer(StringRedisSerializer.UTF_8);template.afterPropertiesSet();RedisConnection redisConnection template.getConnectionFactory().getConnection();redisConnection.subscribe(new MessageListener() {Overridepublic void onMessage(Message message, byte[] bytes) {// 收到消息的处理逻辑System.out.println(Receive message : message);}}, mychannel.getBytes(StandardCharsets.UTF_8));}
}2.1 消息监听容器Message Listener Containers
由于其阻塞特性低级别订阅RedisConnection的订阅没有吸引力因为它需要对每个监听器进行连接和线程管理。为了缓解这个问题Spring Data提供了RedisMessageListenerContainer它完成了所有繁重的工作。如果你熟悉EJB和JMS你应该会发现这些概念很熟悉因为它的设计尽可能地接近Spring Framework的支持及其消息驱动的POJOMDP。 RedisMessageListenerContainer充当消息监听器容器。它用于从Redis通道channel接收消息并驱动注入其中的MessageListener实例。监听器容器负责消息接收的所有线程并将消息分派到监听器中进行处理。消息监听器容器是MDP和消息传递提供者之间的中介负责注册接收消息、资源获取和释放、异常转换等。这使你作为应用程序开发人员能够编写与接收消息相关联的可能复杂的业务逻辑并将Redis基础设施的公式化问题委托给框架。 MessageListener还可以实现SubscriptionListener以便在确认订阅/取消订阅时接收通知。在同步调用时监听订阅通知可能很有用。 为了最大限度地减少应用程序占用RedisMessageListenerContainer允许多个监听器共享一个连接和一个线程即使它们不共享订阅。因此无论应用程序跟踪多少监听器或通道运行时成本在其整个生命周期中都保持不变。另外容器允许更改运行时配置以便在应用程序运行时添加或删除监听器而无需重新启动。此外容器使用延迟订阅方法仅在需要时使用RedisConnection。如果所有监听器都被取消订阅则会自动执行清理并释放线程。 为了保证消息的异步特性容器需要一个java.util.concurrent.Executor或Spring的TaskExecutor来分发消息。根据负载、监听器的数量和运行时环境你应该调整executor以更好地满足你的需求。
2.2 消息监听适配器The MessageListenerAdapter
MessageListenerAdapter类是Spring异步消息传递支持的最后一个组件。简而言之它允许你将几乎任何类公开为MDP(尽管有一些约束)。以下面的接口定义举例:
public interface MessageDelegate {void handleMessage(String message);void handleMessage(Map message);void handleMessage(byte[] message);void handleMessage(Serializable message);// pass the channel/pattern as wellvoid handleMessage(Serializable message, String channel);}请注意尽管上面的接口没有扩展MessageListener接口但仍然可以通过使用MessageListenerAdapter类将其用作MDP。还请注意各种消息处理方法是如何根据它们可以接收和处理的各种消息类型的内容进行强类型化的。此外消息发送到的通道channel或模式pattern可以作为第二个String类型的参数传递给方法:
public class DefaultMessageDelegate implements MessageDelegate {// implementation elided for clarity...
}注意上面的MessageDelegate接口的实现(上面的DefaultMessageDelegate类)是如何完全没有Redis依赖的。它确实是一个POJO我们使用以下配置将其创建为MDP:
Configuration
class MyConfig {// …BeanDefaultMessageDelegate listener() {return new DefaultMessageDelegate();}BeanMessageListenerAdapter messageListenerAdapter(DefaultMessageDelegate listener) {return new MessageListenerAdapter(listener, handleMessage);}BeanRedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory, MessageListenerAdapter listener) {RedisMessageListenerContainer container new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.addMessageListener(listener, ChannelTopic.of(chatroom));return container;}
}监听器主题可以是一个通道例如topic“chatroom”也可以是一种模式例如topic“*room” 前面的示例使用Redis命名空间来声明消息监听器容器并自动将POJO注册为监听器。成熟beans的定义如下
bean idmessageListener classorg.springframework.data.redis.listener.adapter.MessageListenerAdapterconstructor-argbean classredisexample.DefaultMessageDelegate//constructor-arg
/beanbean idredisContainer classorg.springframework.data.redis.listener.RedisMessageListenerContainerproperty nameconnectionFactory refconnectionFactory/property namemessageListenersmapentry key-refmessageListenerbean classorg.springframework.data.redis.listener.ChannelTopicconstructor-arg valuechatroom//bean/entry/map/property
/bean每次接收到消息时适配器都会自动且透明地在低级(low-level)格式和所需的对象类型之间执行转换(使用配置的RedisSerializer)。由方法调用引起的任何异常都将被捕获并由容器处理(默认情况下异常将被记录)。
三、反应式消息监听器容器Reactive Message Listener Container
Spring Data提供了ReactiveRedisMessageListenerContainer它帮助用户完成所有繁重的转换和订阅状态管理工作。 消息监听器容器本身不需要外部线程资源。它使用driver线程来发布消息。
ReactiveRedisConnectionFactory factory …
ReactiveRedisMessageListenerContainer container new ReactiveRedisMessageListenerContainer(factory);FluxChannelMessageString, String stream container.receive(ChannelTopic.of(my-channel));要等待并确保正确的订阅可以使用receiveLater方法该方法返回MonoFlux。返回的Mono与内部发布者一起完成作为完成对给定主题的订阅的结果。通过拦截onNext信号你可以同步服务器端订阅。
ReactiveRedisConnectionFactory factory …
ReactiveRedisMessageListenerContainer container new ReactiveRedisMessageListenerContainer(factory);MonoFluxChannelMessageString, String stream container.receiveLater(ChannelTopic.of(my-channel));stream.doOnNext(inner - // notification hook when Redis subscriptions are synchronized with the server).flatMapMany(Function.identity()).…;3.1 通过template API订阅Subscribing via template API
如上所述你可以直接使用ReactiveRedisTemplate订阅通道/模式。这种方法提供了一种直接但有限的解决方案因为你失去了在初始订阅之后添加订阅的选项。尽管如此你仍然可以通过返回的Flux来控制消息流例如使用take(Duration)。当读取完成、出错或取消时所有绑定的资源将再次释放。
redisTemplate.listenToChannel(channel1, channel2).doOnNext(msg - {// message processing ...
}).subscribe();
