SpringBoot+Redis实现消息的发布与订阅的示例代码

本文主要介绍了SpringBoot+Redis实现消息的发布与订阅，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

1.什么是redis的发布与订阅

在官网的文档介绍中有一行介绍：Redis是一个快速稳定的发布/订阅消息系统。

2.Redis发布订阅

机制

Redis提供了发布与订阅的功能，可以用于消息的传输，Redis的发布订阅机制包括三部分，发布者、订阅者和Channel（主题或者队列）。

3.命令行实现功能

订阅主题

Redis采用SUBSCRIBE channel命令订阅某个主题，返回的参数subscribe表示已经成功订阅主题，第二个参数表示订阅的主题名称，第三个表示当前订阅数

127.0.0.1:6379> SUBSCRIBE ORDER-PAY-SUCCESS Reading messages... (press Ctrl-C to quit) 1) "subscribe" 2) "ORDER-PAY-SUCCESS" 3) (integer) 1

模式匹配订阅

模式匹配功能是允许客户端订阅匹配某种模式的频道，Redis采用PSUBSCRIBE channel命令订阅匹配某种模式的所有频道，用*表示模式，*可以被任意值代替。

127.0.0.1:6379> PSUBSCRIBE ORDER-PAY-* Reading messages... (press Ctrl-C to quit) 1) "psubscribe" 2) "ORDER-PAY-*" 3) (integer) 1

发布消息

Redis采用PUBLISH channel message命令在某个主题上发布消息，返回的参数是接收到消息的订阅者个数

127.0.0.1:6379> PUBLISH ORDER-PAY-SUCCESS DD202109071525 (integer) 1

取消订阅

Redis采用UNSUBSCRIBE channel或PUNSUBSCRIBE channel命令取消某个主题的订阅，由于Redis的订阅操作是阻塞式的，因此一旦客户端订阅了某个频道或模式，就将会一直处于订阅状态直到退出。在 SUBSCRIBE，PSUBSCRIBE，UNSUBSCRIBE 和 PUNSUBSCRIBE 命令中，其返回值都包含了该客户端当前订阅的频道和模式的数量，当这个数量变为0时，该客户端会自动退出订阅状态。

127.0.0.1:6379> UNSUBSCRIBE ORDER-PAY-SUCCESS 1) "unsubscribe" 2) "ORDER-PAY-SUCCESS" 3) (integer) 0 127.0.0.1:6379> PUNSUBSCRIBE ORDER-PAY-SUCCESS 1) "punsubscribe" 2) "ORDER-PAY-SUCCESS" 3) (integer) 0

测试

首先三个订阅者订阅ORDER-PAY-SUCCESS主题，当发送者在这个主题内发送订单号时，所有的订阅者都会收到这个主题的消息。

4.SpringBoot实现功能

Springboot整合Redis

引入依赖包

  org.springframework.bootspring-boot-starter2.2.10.RELEASE org.springframework.bootspring-boot-starter-web2.2.10.RELEASE org.springframework.bootspring-boot-starter-data-redis2.4.1

配置yaml文件

server:   port: 8888 spring:   application:     name: spring-boot-redis   redis:     host: 127.0.0.1     port: 6379

配置消息监听

import com.demo.redis.listener.MessageListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /**  * @author Bai  * @date 2021/4/29 上午 11:17  * @description  */ @Configuration public class RedisMessageConfig {     /**      * 监听订单支付完成主题      */     private static final String ORDER_PAY_SUCCESS = "ORDER-PAY-SUCCESS";     /**      * 注入消息监听适配器      * @param messageListener      * @return {@link MessageListenerAdapter}      * @throws      * @author Bai      * @date 2021/9/7 下午 03:54      */     @Bean     public MessageListenerAdapter getMessageListenerAdapter(MessageListener messageListener){         return new MessageListenerAdapter(messageListener, "onMessage");     }     /**      * 注入消息监听容器      * @param redisConnectionFactory      * @param messageListenerAdapter      * @return {@link RedisMessageListenerContainer}      * @throws      * @author Bai      * @date 2021/9/7 下午 03:54      */     @Bean     public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, MessageListenerAdapter messageListenerAdapter){         RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();         redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);         //订阅订单支付成功主题         redisMessageListenerContainer.addMessageListener(messageListenerAdapter, new PatternTopic(ORDER_PAY_SUCCESS));         return redisMessageListenerContainer;     }     /**      * 处理内容      * @param connectionFactory      * @return {@link StringRedisTemplate}      * @throws      * @author Bai      * @date 2021/9/7 下午 04:01      */     @Bean     StringRedisTemplate template(RedisConnectionFactory connectionFactory) {         return new StringRedisTemplate(connectionFactory);     } }

接收消息

import org.springframework.stereotype.Component; /**  * @author Bai  * @date 2021/9/7 下午 03:51  * @description  */ @Component public class MessageListener {     public void onMessage(String message){         System.out.println("接收消息：" + message);     } }

测试

发送订单支付成功的消息

import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.test.context.junit4.SpringRunner; import javax.annotation.Resource; /**  * @author Bai  * @date 2021/9/7 下午 04:32  * @description  */ @RunWith(value = SpringRunner.class) @SpringBootTest public class RedisPubTest {     /**      * 订单支付完成主题      */     private static final String ORDER_PAY_SUCCESS = "ORDER-PAY-SUCCESS";     @Resource     private StringRedisTemplate stringRedisTemplate;     /**      * 模拟发送5调订单支付完成的消息      * @throws      * @author Bai      * @date 2021/9/7 下午 04:57      */     @Test     public void sendMessage(){         for (int i = 0; i <5; i++) {             stringRedisTemplate.convertAndSend(ORDER_PAY_SUCCESS,"DD" + System.currentTimeMillis());         }     } }

接收到订单支付成功的消息

. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.2.10.RELEASE)
2021-09-07 16:56:54.729 INFO 3712 --- [ main] com.demo.redis.RedisSubPubApplication : Starting RedisSubPubApplication on DESKTOP-595LI4G with PID 3712 (D:\Bai\Sources\spring-boot-demo\redis-sub-pub\target\classes started by 1 in D:\Bai\Sources\spring-boot-demo)
2021-09-07 16:56:54.735 INFO 3712 --- [ main] com.demo.redis.RedisSubPubApplication : No active profile set, falling back to default profiles: default
2021-09-07 16:56:55.205 INFO 3712 --- [ main] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode!
2021-09-07 16:56:55.207 INFO 3712 --- [ main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data Redis repositories in DEFAULT mode.
2021-09-07 16:56:55.229 INFO 3712 --- [ main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 11 ms. Found 0 Redis repository interfaces.
2021-09-07 16:56:55.557 INFO 3712 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 8888 (http)
2021-09-07 16:56:55.565 INFO 3712 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat]
2021-09-07 16:56:55.565 INFO 3712 --- [ main] org.apache.catalina.core.StandardEngine : Starting Servlet engine: [Apache Tomcat/9.0.38]
2021-09-07 16:56:55.669 INFO 3712 --- [ main] o.a.c.c.C.[Tomcat].[localhost].[/] : Initializing Spring embedded WebApplicationContext
2021-09-07 16:56:55.669 INFO 3712 --- [ main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 895 ms
2021-09-07 16:56:56.032 INFO 3712 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor'
2021-09-07 16:56:56.950 INFO 3712 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 8888 (http) with context path ''
2021-09-07 16:56:56.952 INFO 3712 --- [ main] com.demo.redis.RedisSubPubApplication : Started RedisSubPubApplication in 2.557 seconds (JVM running for 3.436)
接收消息：DD1631005025949
接收消息：DD1631005025964
接收消息：DD1631005025965
接收消息：DD1631005025967
接收消息：DD1631005025968