SpringBoot+Redis實現訊息的釋出與訂閱
SpringBoot+Redis實現訊息的釋出與訂閱
1.什麼是redis的釋出與訂閱
在官網的文件介紹中有一行介紹:Redis是一個快速穩定的釋出/訂閱訊息系統。
2.Redis釋出訂閱
機制
Redis提供了釋出與訂閱的功能,可以用於訊息的傳輸,Redis的釋出訂閱機制包括三部分,釋出者、訂閱者和Channel(主題或者佇列)。
3.命令列實現功能
訂閱主題
Redis採用SUBSCRIBE channel
命令訂閱某個主題,返回的引數subscribe表示已經成功訂閱主題,第二個引數表示訂閱的主題名稱,第三個表示當前訂閱數
shell
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
命令訂閱匹配某種模式的所有頻道,用*表示模式,*可以被任意值代替。
shell
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
命令在某個主題上釋出訊息,返回的引數是接收到訊息的訂閱者個數
shell
127.0.0.1:6379> PUBLISH ORDER-PAY-SUCCESS DD202109071525
(integer) 1
取消訂閱
Redis採用UNSUBSCRIBE channel
或PUNSUBSCRIBE channel
命令取消某個主題的訂閱,由於Redis的訂閱操作是阻塞式的,因此一旦客戶端訂閱了某個頻道或模式,就將會一直處於訂閱狀態直到退出。在 SUBSCRIBE
,PSUBSCRIBE
,UNSUBSCRIBE
和 PUNSUBSCRIBE
命令中,其返回值都包含了該客戶端當前訂閱的頻道和模式的數量,當這個數量變為0時,該客戶端會自動退出訂閱狀態。
shell
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
引入依賴包
xml
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
<version>2.2.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.2.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<version>2.4.1</version>
</dependency>
</dependencies>
配置yaml檔案
```yaml server: port: 8888
spring: application: name: spring-boot-redis redis: host: 127.0.0.1 port: 6379 ```
配置訊息監聽
```java 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);
}
} ```
接收訊息
```java 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);
}
} ```
測試
傳送訂單支付成功的訊息
```java 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());
}
}
} ```
接收到訂單支付成功的訊息
```shell
. _ _ _ /\ / ' _ () _ \ \ \ \ ( ( )___ | ' | '| | ' \/ ` | \ \ \ \ \/ )| |)| | | | | || (| | ) ) ) ) ' |_| .|| ||| |_, | / / / / =========|_|==============|/=//// :: 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 ```