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

淩亂°似流年 2022-08-23 03:52 97阅读 0赞

### 今天来学习如何利用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主题发送一个消息。我们看到，消息可以被成功的接收到并打印出来，搞定！