Spring Boot使用Redis进行消息的发布订阅

时间:2023-12-19 14:02:44

今天来学习如何利用Spring Data对Redis的支持来实现消息的发布订阅机制。发布订阅是一种典型的异步通信模型,可以让消息的发布者和订阅者充分解耦。在我们的例子中,我们将使用StringRedisTemplate来发布一个字符串消息,同时基于MessageListenerAdapter使用一个POJO来订阅和响应该消息。

提示

事实上,RedisRedis 不仅提供一个NoSQL数据库,同时提供了一套消息系统。

环境准备

开发环境:

  • IDE+Java环境(JDK 1.7或以上版本)
  • Maven 3.0+(Eclipse和Idea IntelliJ内置,如果使用IDE并且不使用命令行工具可以不安装)

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <groupId>com.tianmaing</groupId>
<artifactId>redis-message</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging> <name>redis-message</name>
<description>Demo of message processing by redis</description> <parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.2.5.RELEASE</version>
<relativePath/>
</parent> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties> <dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
</dependency>
</dependencies> <build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build> </project>

通过配置spring-boot-starter-redis依赖,把Spring Boot对Redis的相关支持引入进来。

创建Redis消息的接收者

在任何一个基于消息的应用中,都有消息发布者和消息接收者(或者称为消息订阅者)。创建消息的接收者,我们只需一个普通POJO,在POJO中定义一个接收消息的方法即可:

package com.tianmaying.springboot.redisdemo;

import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired; public class Receiver {
private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class); private CountDownLatch latch; @Autowired
public Receiver(CountDownLatch latch) {
this.latch = latch;
} public void receiveMessage(String message) {
LOGGER.info("Received <" + message + ">");
latch.countDown();
}
}

这个Receiver类将会被注册为一个消息监听者时。处理消息的方法我们可以任意命名,我们有相当大的灵活性。

我们给Receiver的构造函数通过@AutoWired标注注入了一个CountDownLatch实例,当接收到消息时,调用cutDown()方法。

注册监听者和发送消息

Spring Data Redis提供基于Redis发送和接收消息的所有需要的组件,我们只需要配置好三个东西:

  • 一个连接工厂(connection factory)
  • 一个消息监听者容器(message listener container)
  • 一个Redis的模板(redis template)

我们将通过Redis模板来发送消息,同时将Receiver注册给消息监听者容器。连接工厂将两者连接起来,使得它们可以通过Redis服务器通信。如何连接呢? 我们将连接工厂实例分别注入到监听者容器和Redis模板中即可。

package com.tianmaying.springboot.redisdemo;

import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
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; @SpringBootApplication
public class App { private static final Logger LOGGER = LoggerFactory.getLogger(App.class); @Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.addMessageListener(listenerAdapter, new PatternTopic("chat")); return container;
} @Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
return new MessageListenerAdapter(receiver, "receiveMessage");
} @Bean
Receiver receiver(CountDownLatch latch) {
return new Receiver(latch);
} @Bean
CountDownLatch latch() {
return new CountDownLatch(1);
} @Bean
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
return new StringRedisTemplate(connectionFactory);
} public static void main(String[] args) throws InterruptedException { ApplicationContext ctx = SpringApplication.run(App.class, args); StringRedisTemplate template = ctx.getBean(StringRedisTemplate.class);
CountDownLatch latch = ctx.getBean(CountDownLatch.class); LOGGER.info("Sending message...");
template.convertAndSend("chat", "Hello from Redis!"); latch.await(); System.exit(0);
}
}

连接工程我们使用Spring Boot默认的RedisConnectionFactory,是Jedis Redis库提供的JedisConnectionFactory实现。

我们将在listenerAdapter方法中定义的Bean注册为一个消息监听者,它将监听chat主题的消息。

因为Receiver类是一个POJO,要将它包装在一个消息监听者适配器(实现了MessageListener接口),这样才能被监听者容器RedisMessageListenerContainer的addMessageListener方法添加到连接工厂中。有了这个适配器,当一个消息到达时,就会调用receiveMesage()`方法进行响应。

就这么简单,配置好连接工厂和消息监听者容器,你就可以监听消息啦!

发送消息就更简单了,我们使用StringRedisTemplate来发送键和值均为字符串的消息。在main()方法中我们创建一个Spring应用的Context,初始化消息监听者容器,开始监听消息。然后获取StringRedisTemplate的实例,往chat主题发送一个消息。我们看到,消息可以被成功的接收到并打印出来,搞定!