SpringBoot 整合Redis 实现发布/订阅模式附带Redis集群配置

一、前言

其实光从代码层面上讲，其实没有什么变化，主要是变化是关于Redis的配置需要更改为集群配置而已，之前接触过redis的话，那么就只需要看一下redis集群配置文件即可了。

对redis实现发布/订阅感兴趣的话，那就可以接着看下去了哈。

发布/订阅模式：所谓发布/订阅模式，其实就是和你关注微信公众号一样的意思。

举个例子：你订阅了两个微信公众号（一个叫青年湖南，一个叫央视新闻），假如我也订阅了青年湖南，某一天央视发布了一条新新闻，你能收到，我没有关注，则我不能收到。但是某周看青年大学习发布王冰冰叫你去学习时，你我都订阅了，就都可以收到。

二、前期准备

两份配置文件都有。

单机也是可以的，想一起搭集群玩的可以👉Docker搭建Redis Cluster 集群环境。

2.1、项目结构：

2.2、依赖的jar包

我这里是因为是习惯创建maven项目，然后将SpringBoot的版本抽出来，方便控制版本。

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.5.2</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
        <version>2.4.3</version>
    </dependency>
     <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
        <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.72</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.20</version>
    </dependency>
</dependencies>

2.3 、yml配置文件

单机配置文件

spring:
  redis:
    database: 0
    port: 6379
    host: localhost
    password:
    lettuce:
      pool:
        # 连接池最大连接数（使用负值表示没有限制）
        max-active: 1024
        # 连接池最大阻塞等待时间（使用负值表示没有限制）
        max-wait: 10000
        # 连接池中的最大空闲连接
        max-idle: 200
        # 连接池中的最小空闲连接
        min-idle: 0
    # 连接超时时间（毫秒）
    timeout: 10000

redis集群配置文件

server:
  port: 8089
spring:
  application:
    name: springboot-redis
  redis:
    password: 1234
    cluster:
      nodes:
        - IP地址:6379
        - IP地址:6380
        - IP地址:6381
        - IP地址:6382
        - IP地址:6383
        - IP地址:6384
      max-redirects: 3  # 获取失败 最大重定向次数
    lettuce:
      pool:
        max-active: 1000  #连接池最大连接数（使用负值表示没有限制）
        max-idle: 10 # 连接池中的最大空闲连接
        min-idle: 5 # 连接池中的最小空闲连接

#===========jedis配置方式=============================================
#    jedis:
#      pool:
#        max-active: 1000  # 连接池最大连接数（使用负值表示没有限制）
#        max-wait: -1ms      # 连接池最大阻塞等待时间（使用负值表示没有限制）
#        max-idle: 10      # 连接池中的最大空闲连接
#        min-idle: 5       # 连接池中的最小空闲连接
#

三、编码

3.1、config层

redis配置类

import com.crush.ps.subscribe.AConsumerRedisListener;
import com.crush.ps.subscribe.BConsumerRedisListener;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
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.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
 * redis 配置类
 * 1. 设置RedisTemplate序列化/返序列化
 * 2. 监听消息
 * @author cuberxp
 * @since 1.0.0
 * Create time 2020/1/23 0:06
 */
@Configuration
@ConditionalOnClass(RedisOperations.class)
@EnableConfigurationProperties(RedisProperties.class)
public class RedisConfig {

    @Autowired
    AConsumerRedisListener aConsumerRedisListener;

    @Autowired
    BConsumerRedisListener bConsumerRedisListener;

    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        //将消息侦听器添加到（可能正在运行的）容器中。 如果容器正在运行，则侦听器会尽快开始接收（匹配）消息。
        // a 订阅了 topica、topicb 两个 频道
        container.addMessageListener(aConsumerRedisListener, new PatternTopic("topica"));
        container.addMessageListener(aConsumerRedisListener, new PatternTopic("topicb"));

        // b 只订阅了 topicb  频道
        container.addMessageListener(bConsumerRedisListener, new PatternTopic("topicb"));
        return container;
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        //设置value hashValue值的序列化
        Jackson2JsonRedisSerializer<Object> serializer = new Jackson2JsonRedisSerializer<Object>(
                Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        serializer.setObjectMapper(om);
        redisTemplate.setValueSerializer(serializer);
        redisTemplate.setHashValueSerializer(serializer);
        //key hasKey的序列化
        redisTemplate.setKeySerializer(stringRedisSerializer);
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

}

3.2、订阅者

我在这边写了两个订阅者,方便演示例子罢了。

A消费者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

/**
 * @author crush
 * MessageListener ： Redis中发布的消息的侦听器。
 */
@Component
public class AConsumerRedisListener implements MessageListener {

    @Autowired
    private  RedisTemplate<String, Object> redisTemplate;

     /**
     * @param message 传递过来的信息数据
     * @param pattern 频道
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        doBusiness(message);
    }

    /**
     * 打印 message body 内容
     *
     * deserialize 从给定的二进制数据反序列化一个对象。
     * @param message
     */
    public void doBusiness(Message message) {
        Object value = redisTemplate.getValueSerializer().deserialize(message.getBody());
        System.out.println("A==>consumer message: " + value.toString());
    }

}

B消费者：

package com.crush.ps.subscribe;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

/**
 * @author crush
 */
@Component
public class BConsumerRedisListener implements MessageListener {

    @Autowired
    private  RedisTemplate<String, Object> redisTemplate;

    /**
     * @param message 传递过来的信息数据
     * @param pattern 频道
     */
    @Override
    public void onMessage(Message message, byte[] pattern) {
        doBusiness(message);
    }

    /**
     * 打印 message body 内容
     *
     * deserialize 从给定的二进制数据反序列化一个对象。
     * @param message
     */
    public void doBusiness(Message message) {
        Object value = redisTemplate.getValueSerializer().deserialize(message.getBody());
        System.out.println("B==>consumer message: " + value.toString());
    }
}

3.3、AnnouncementMessage实体类

就是自己写的传递消息的实体类，（AnnouncementMessage 意思就是拿来模拟发布公布的实体类）

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;

/**
 * @author crush
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class AnnouncementMessage implements Serializable {

    private static final long serialVersionUID = 8632296967087444509L;

    /**
     * 公告信息id
     */
    private String id;

    /**
     * 公告内容
     */
    private String content;
}

四、测试

@SpringBootTest
public class SubscribeTest {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Test
    public void testSubscribe() {
        String achannel = "topica";
        String bchannel = "topicb";
        
        redisTemplate.convertAndSend(achannel, "hello world");
        
        redisTemplate.convertAndSend(bchannel, new AnnouncementMessage("1", "模拟发通告"));
    }
}

结果：

输出： 
A==>consumer message: hello world
A==>consumer message: AnnouncementMessage(id=1, content=模拟发通告)

B==>consumer message: AnnouncementMessage(id=1, content=模拟发通告)

因为A 消费者订阅两个频道，而B 消费者只订阅了一个频道，所以A 会多一条。

注:测试时需要把主启动类也给启动起来，方便查看输出。（主启动自己写就好了，没有什么其他的注解，普普通通的）