[SpingBoot guides系列翻译]Redis的消息订阅发布

时间：2023-03-09 22:16:06
[SpingBoot guides系列翻译]Redis的消息订阅发布

Redis的消息

部分参考链接

原文

CountDownLatch

概述

目的

这节讲的是用Redis来实现消息的发布和订阅，这里会使用Spring Data Redis来完成。

这里会用到两个东西，StringRedisTemplate和MessageListenerAdapter。分别用来发布String类型的消息和订阅接收这些消息。

你需要的准备的

  • 大概15min(实际用下来应该不够)
  • 喜欢的ide或者文本编辑器(我使用intellij)
  • Jdk1.8+
  • Gradle4+ 或者 Maven3.2+(这里用Maven)

如何通过Maven来完成

新建maven项目

[SpingBoot guides系列翻译]Redis的消息订阅发布

创建pom.xml文件

<?xml version="1.0" encoding="UTF-8"?>

<!--

  ~ Copyright (c) 2019.

  ~ lou

  -->

<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>org.springframework</groupId>

    <artifactId>gs-messaging-redis</artifactId>

    <version>1.0-SNAPSHOT</version>

    <parent>

        <groupId>org.springframework.boot</groupId>

        <artifactId>spring-boot-starter-parent</artifactId>

        <version>2.1.4.RELEASE</version>

    </parent>

    <properties>

        <java.version>1.8</java.version>

    </properties>

    <dependencies>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter</artifactId>

            <version>2.1.4.RELEASE</version>

        </dependency>

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-data-redis</artifactId>

            <version>1.5.2.RELEASE</version>

        </dependency>

    </dependencies>

    <build>

        <plugins>

            <plugin>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-maven-plugin</artifactId>

            </plugin>

        </plugins>

    </build>

</project>

Spring Boot Maven Plugin 有以下几个作用

  • 把项目打包成一个可执行的jar文件。
  • 搜索并标记public static void main()为可执行类。
  • 使用内置的依赖管理

需要一个Redis Server

这里还需要一个Redis Server，搭设过程就不演示了，假设已经搭好，server: localhost:6379, password:xxxxxx。

创建一个Redis 消息接收类

src/main/java/hello/Receiver.java

/*

 * Copyright (c) 2019.

 * lou

 */

package hello;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.CountDownLatch;

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("接收到消息"+message);

        latch.countDown();

    }

}

CountDownLatch用来阻塞当前进程，调用countDown方法一次减少1，到0释放。。CountDownLatch 无法重置，如有需要，使用CyclicBarrier。

创建一个Redis配置类

Redis的配置应该是放在配置文件中的，所以需要创建一个redis配置类，来读取resource/application.properties中的配置并赋值给相应的connectionFactory。

先添加一下application.properties

spring.redis.password=xxxxx

spring.redis.port=6379

spring.redis.host=localhost

然后创建一个hello/RedisStandaloneConfigurationPropertySource.java

/*

 * Copyright (c) 2019.

 * lou

 */

package hello;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.context.annotation.PropertySource;

import org.springframework.data.redis.connection.RedisConnectionFactory;

import org.springframework.data.redis.connection.RedisStandaloneConfiguration;

import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;

//配置RedisConnectionFactory

@Configuration

@PropertySource(value = "application.properties",encoding = "utf-8")

public class RedisStandaloneConfigurationPropertySource {

    private Logger logger = LoggerFactory.getLogger(RedisStandaloneConfigurationPropertySource.class);

    @Value("${spring.redis.host}")

    private String hostName;

    @Value("${spring.redis.port}")

    private int port;

    @Value("${spring.redis.password}")

    private String password;

    @Override

    public String toString() {

        return "RedisStandaloneConfigurationPropertySource{" +

                "hostName='" + hostName + '\'' +

                ", port=" + port +

                ", password='" + password + '\'' +

                '}';

    }

    @Bean

    RedisConnectionFactory getRedisConnectionFactory() {

      	//日志记录一下

        logger.info("redis的配置为："+this.toString());

        JedisConnectionFactory factory = new JedisConnectionFactory();

        RedisStandaloneConfiguration configuration = factory.getStandaloneConfiguration();

        configuration.setHostName(hostName);

        configuration.setPort(port);

        configuration.setPassword(password);

        return factory;

    }

}

这个类的功能就是读取配置文件，并且定义一个RedisConnectionFactory的Bean，后面会用到。

注册这个消息接收类并发送消息

Spring Data Redis提供了发送和接收消息的所有组件。重点需要配置以下几个东西

  • 一个Connection Factory(连接管理工厂)
  • 一个 Message Listener Container(消息接收容器)
  • 一个Redis Template

使用Redis Template来发送消息，然后把Receiver注册到Message Listener Container中。Connection Factory驱动Template和Container，让他们可以连接到redis server。

例子中使用Spring boot的RedisConnectionFactory，它是基于Jedis的JedisConnectionFactory的实例。Connection Factory会同时注入到container和template中。

src/main/java/hello/Application.java

/*

 * Copyright (c) 2019.

 * lou

 */

package hello;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Qualifier;

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.connection.jedis.JedisConnectionFactory;

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;

import java.util.concurrent.CountDownLatch;

@SpringBootApplication

public class Application {

    private static final Logger logger = LoggerFactory.getLogger(Application.class);

    //Container 由RedisConnectionFactory 和MessageListenerAdapter实例化

    @Bean

    RedisMessageListenerContainer container(@Qualifier("getRedisConnectionFactory") RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();

        container.setConnectionFactory(connectionFactory);

        container.addMessageListener(listenerAdapter, new PatternTopic("chat"));

        return container;

    }

    @Bean

    MessageListenerAdapter listenerAdapter(Receiver receiver) {

        //通过Receiver实例化Adapter，设置默认listen方法，receiveMessage

        return new MessageListenerAdapter(receiver, "receiveMessage");

    }

    @Bean

    Receiver receiver(CountDownLatch latch) {

        return new Receiver(latch);

    }

    @Bean

    CountDownLatch latch() {

        return new CountDownLatch(100);//设置上限100

    }

    //StringRedisTemplate由RedisConnectionFactory实例化

    @Bean

    StringRedisTemplate template(@Qualifier("getRedisConnectionFactory") RedisConnectionFactory connectionFactory) {

        return new StringRedisTemplate(connectionFactory);

    }

    public static void main(String[] args) throws InterruptedException {

        ApplicationContext ctx = SpringApplication.run(Application.class, args);

        CountDownLatch latch = ctx.getBean(CountDownLatch.class);

        latch.await();

        System.exit(0);

    }

}

这样Application就完成了。

  • RedisMessageListenerContainer由container方法返回。
  • MessageListenerAdapter 由listenAdapter方法返回，构造函数里面第二个参数表示调用的方法receiveMessage，这个在receiver类中已经定义。
  • StringRedisTemplate 由template方法返回，这个类关注于操作key和value都是String类型的数据。

创建一个webapi来测试这个功能

src/main/java/hello/controller/TestController.java

/*

 * Copyright (c) 2019.

 * lou

 */

package hello.controller;

import org.springframework.data.redis.core.StringRedisTemplate;

import org.springframework.data.redis.core.ValueOperations;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RequestParam;

import org.springframework.web.bind.annotation.RestController;

import java.util.Random;

@RestController

public class TestController {

    Random random = new Random();

    StringRedisTemplate template;

    public TestController(StringRedisTemplate template) {

        this.template = template;

    }

    @GetMapping("test")

    public int TestMethod(@RequestParam(name = "t", defaultValue = "hello") String t) {

        ValueOperations operations = template.opsForValue();

        //测试读写redis

      	operations.set("t", t);

      	//发送消息

        template.convertAndSend("chat", "消息" + t);

        return random.nextInt(100);

    }

}

test方法里面调用convertAndSend方法，发送消息，启动后访问http://localhost:8080/test?t=2432，可以看到日志上有返回内容。

[SpingBoot guides系列翻译]Redis的消息订阅发布

构建可执行jar

打包 mvn package。

运行 java -jar ./target/xxx.jar。看到可以运行起来。

小结

这个demo教了我们如何通过配置文件配置redis连接，然后实现发布和订阅。

相关文章