本篇内容主要讲解“SpringBoot Redis发布订阅模式的方法是什么”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“SpringBoot Redis发布订阅模式的方法是什么”吧!
注意:redis的发布订阅模式不可以将消息进行持久化,订阅者发生网络断开、宕机等可能导致错过消息。
发布者通过以下命令可以往指定channel发布message
redis> publish channel message
订阅者通过以下命令可以订阅一个或多个频道,如果频道不存在则会创建
redis> subscribe channel [channel ...]
对于redis的发布订阅的命令就这么简单。那么接下来我们在springboot中如何使用发布订阅的功能呢?
添加依赖配置redis信息和连接池什么的就不说了,如果添加的有commons-pool2依赖的话,会自动帮我们配置redis连接池的
相对于订阅者来说,发布者的实现方式很简单,以下方式就可以往channel中发送message了。
@Resource private RedisTemplate<String, Object> redisTemplate; public void publish(){ // 使用高级的redisTemplate redisTemplate.convertAndSend("channel","message"); // 使用低级的connection 实际上redisTemplate的底层就是使用的下面的方式 redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { connection.publish("channel".getBytes(StandardCharsets.UTF_8), "message".getBytes(StandardCharsets.UTF_8)); return null; } }, true); // true这个参数意思是 是否将redis连接暴露给回调代码,大多数情况下设置true就可以了,往后深入的话可以看到 RedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse)); 如果为false的话会创建redis连接的代理 }
订阅者因为涉及到连接、线程等 所以内容相对会多一点
@Resource private RedisTemplate<String, Object> redisTemplate; public void subscribe() { redisTemplate.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { // 我定义了一个全局的 ConcurrentHashMap 用来存放连接 因为后面的取消订阅的线程要和订阅的线程用同一个连接 map.put("connection",connection); // subscribe 按频道订阅 该方法会阻塞该线程 只有取消订阅才会释放该线程 connection.subscribe(new MessageListener() { @Override public void onMessage(Message message, byte[] pattern) { log.info("接收到消息"); System.out.println(new String(message.getBody())); } }, "channelOne".getBytes(StandardCharsets.UTF_8), "channelTwo".getBytes(StandardCharsets.UTF_8)); // 按模式订阅 pSubscribe 只有取消订阅才会释放该线程 // connection.pSubscribe(new MessageListener() { // @Override // public void onMessage(Message message, byte[] pattern) { // System.out.println(new String(message.getBody())); // } // }, "patternOne".getBytes(StandardCharsets.UTF_8), "patternOne".getBytes(StandardCharsets.UTF_8)); return null; } }, true); }
如何取消订阅呢?从刚才的map里取到连接
RedisConnection the = map.get("connection"); Subscription subscription = the.getSubscription(); subscription.unsubscribe();
上面的那种订阅为低级订阅,由于连接在调用subscribe的时候会导致当前线程阻塞,这种方式需要对每个监听器连接和线程管理,所以spring提供了RedisMessageListenerContainer类来帮我们完成这些工作。
RedisMessageListenerContainer顾名思义可以知道它是一个消息监听容器
详情请参考官方文档
如何实现
@Configuration public class DefaultMessageListenerContainerConfig { @Bean public RedisMessageListenerContainer container(RedisConnectionFactory factory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); // 官方推荐我们使用自定义的线程池或者使用TaskExecutor container.setTaskExecutor(executor()); container.addMessageListener(new MessageListener() { @Override public void onMessage(Message message, byte[] pattern) { System.out.println(Thread.currentThread().getName() + ": " + new String(message.getBody())); } }, new ChannelTopic("message")); return container; } @Bean public TaskExecutor executor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(Runtime.getRuntime().availableProcessors()); executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2); executor.setQueueCapacity(100); executor.initialize(); return executor; } }
这个时候我们在redis命令行内使用 publish channel message 的时候,我们的spring程序就可以订阅到消息了。
再说下 MessageListenerAdapter
我们可以通过 MessageListenerAdapter 消息接收者包装进去,消息接收者不会和redis有任何耦合。
官方文档给了spring传统的xml的方式配置的,下面我给出基于configuration配置的代码
public interface MessageDelegate { void handleMessage(String message); } public class DefaultMessageDelegate implements MessageDelegate { @Override public void handleMessage(String message) { System.out.println(message); } } @Configuration public class MessageListenerContainerConfig { @Autowired private DefaultMessageDelegate defaultMessageDelegate; @Bean public RedisMessageListenerContainer container(RedisConnectionFactory factory, MessageListenerAdapter messageListenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); container.setTaskExecutor(executor()); Map<MessageListenerAdapter, Collection<? extends Topic>> map = new HashMap<>(); List<ChannelTopic> channelTopics = new ArrayList<>(); ChannelTopic channelTopic = new ChannelTopic("message"); channelTopics.add(channelTopic); map.put(messageListenerAdapter, channelTopics); container.setMessageListeners(map); return container; } @Bean public TaskExecutor executor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(Runtime.getRuntime().availableProcessors()); executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2); executor.setQueueCapacity(100); executor.initialize(); return executor; } @Bean public MessageListenerAdapter messageListenerAdapter() { // handleMessage 参数消息来的时候要调用的方法 默认是 handleMessage return new MessageListenerAdapter(defaultMessageDelegate, "handleMessage"); } }
如果我们要在程序运行时添加订阅或者取消订阅的时候该怎么办呢?
我们需要提前准备好消息侦听器,添加的时候把侦听器注入到消息容器
取消的时候就调用消息容器的remove方法把侦听器删除掉即可。
到此,相信大家对“SpringBoot Redis发布订阅模式的方法是什么”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。