Spring Cloud 集成 Redis 发布订阅

前言

在当今的软件开发领域，分布式系统已经成为一种主流的架构模式，尤其是在处理大规模、高并发、高可用的业务场景时。然而，随着系统复杂性的增加，缓存机制的重要性也日益凸显。缓存不仅可以帮助我们减轻数据库压力，提高系统响应速度，还能在一定程度上保证系统的稳定性和可靠性。
Redis，作为一款开源的、高性能的、支持多种数据结构的NoSQL数据库，已经成为缓存领域的佼佼者。它提供了丰富的数据结构支持，如字符串、哈希、列表、集合和有序集合等，使得开发者能够灵活地根据业务需求进行数据存储和读取。同时，Redis还支持事务、持久化、发布订阅等高级功能，极大地增强了其应用场景的广泛性和实用性。
Spring Cloud，作为Spring家族中的一员，致力于提供一套完整的微服务解决方案。它集成了众多的开源组件，使得开发者能够轻松地构建、部署和管理微服务应用。在缓存方面，Spring Cloud也提供了强大的支持，通过集成Redis等缓存技术，可以轻松地实现数据的缓存和共享。
因此，将Spring Cloud与Redis进行集成，不仅可以充分发挥两者的优势，还能为开发者提供更加便捷、高效的缓存解决方案。通过Spring Cloud的集成，我们可以方便地配置Redis缓存的相关参数，如缓存的过期时间、缓存的淘汰策略等。同时，我们还可以利用Spring Cloud提供的分布式缓存机制，实现多个服务节点之间的缓存共享和同步。
在本文中，我们将详细介绍如何在Spring Cloud应用中集成Redis缓存，并展示如何通过简单的配置和代码实现数据的高效缓存和读取。希望通过本文的分享，能够帮助读者更好地理解Spring Cloud与Redis的集成方式，并在实际项目中加以应用，提升系统的性能和稳定性。

步骤

引入相关maven依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

添加相关配置

spring:  
    redis: # Redis相关配置的开始，这里定义了连接到Redis服务所需要的参数  
        host: 127.0.0.1 # Redis服务器的主机地址，这里是本地地址，用于连接本地的Redis服务  
        port: 6379 # Redis服务的端口号，默认通常是6379  
        password: 1q@w3e4r # Redis服务器的密码，如果Redis服务器设置了密码，这里需要填写对应的密码  
        database: 0 # 使用的Redis数据库的索引号，默认为0，Redis可以有多个数据库，通过索引号来区分

在Spring Boot或Spring Cloud应用中，上述配置通常在application.yml或application.properties文件中进行。spring.redis下配置的是连接Redis服务器所必需的参数，包括主机地址、端口号、密码和使用的数据库索引。这些参数将被Spring Boot自动配置机制用于构建Redis连接工厂，从而实现对Redis的访问。
需要注意的是，如果Redis服务器未设置密码，password字段可以省略或留空。同时，如果使用的是默认端口（6379）和默认数据库（0），这些配置项也可以根据实际情况省略。
此外，对于更复杂的Redis使用场景，如连接池配置、哨兵模式或集群模式，还需要添加更多的配置项来指定相关参数。在实际应用中，应根据项目需求和Redis服务器的配置来选择合适的配置项。

使用方法

发布订阅

构建发布订阅全局实体

@Getter
@Setter
public class RedisMessageBo {

    /**
     * 全局唯一id
     */
    private String uuid;

    /**
     * 其他信息
     */
    private String message;
}

声明交换机名称

/**
 * 交换机名称
 *
 * @author 30346
 */
public class ConstantConfiguration {

    public static final String EXCHANGE_NAME = "exchange_name";
}

创建消息适配器

@Configuration
public class RedisListenerAdapter {

    /**
     * 消息适配器
     *
     * @param receiver 接收者
     * @return {@link MessageListenerAdapter}
     */
    @Bean
    MessageListenerAdapter listenerAdapter(MessageSubListener receiver) {
        return new MessageListenerAdapter(receiver);
    }
}

发布监听配置

/**
 * 发布监听配置
 *
 * @author 30346
 */
@Configuration
public class RedisPubListenerConfig {

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter messageListenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 可以添加多个 messageListener，配置不同的交换机
        container.addMessageListener(messageListenerAdapter, new ChannelTopic(ConstantConfiguration.EXCHANGE_NAME));
        return container;
    }
}

订阅监听配置

/**
 * 订阅监听配置
 *
 * @author 30346
 */
@Slf4j
@Component
@Getter
@Setter
@Primary
public abstract class MessageSubListener implements MessageListener {

    private String channel;

    private String msg;

    @Autowired
    private RedisUtil redisUtil;

    //该抽象方法是为了在微服务中多个项目可以使用抽离出来的redis发布订阅功能
    //其他不服中直接集成MessageSubListener并且重写onMessage即可
    public abstract void onMessage(String channel, String msg);

    @Override
    public void onMessage(Message message, byte[] bytes) {
        channel = new String(bytes);
        RedisMessageBo redisMessageBo = JSONUtil.toBean(message.toString(), RedisMessageBo.class);
        msg = redisMessageBo.getMessage();
        //全局唯一id，为了解决集群下消息重复消费问题，我们在redis中存储了全局唯一id作为锁，过期将自动清楚
        String key = channel.concat(":").concat(redisMessageBo.getUuid());
        if (!redisUtil.hasKey(key)) {
            redisUtil.setEx(key, redisMessageBo.getMessage(), 60, TimeUnit.SECONDS);
            onMessage(channel, msg);
        }
    }
}

其他服务订阅监听配置

/**
 * 订阅监听配置
 *
 * @author 30346
 */
@Slf4j
@Component
public class DefaultMessageSubListener extends MessageSubListener {

    @Override
    public void onMessage(String channel, String msg) {
        //业务逻辑
    }
}

发布一个消息

//初始化对象并设置全局唯一id和消息
RedisMessageBo redisMessageBo = new RedisMessageBo();
redisMessageBo.setUuid(UUID.randomUUID().toString(true));
redisMessageBo.setMessage(JSONUtil.toJsonStr(messageBo));
将消息发布到交换机上
stringRedisTemplate.convertAndSend(ConstantConfiguration.EXCHANGE_NAME, JSONUtil.toJsonStr(redisMessageBo));

注意

我们为了抽离发布订阅配置，默认创建了一个空的订阅监听器（保证在其他服务中不需要发布订阅是还可以正常启动），如果其他服务需要实现发布订阅则集成MessageSubListener实现一个订阅监听，并重写onMessage扩展业务逻辑。还有一点要关注一下，@Primary，由于默认创建了一个空的订阅监听器，当其他服务创建了自定义订阅监听器需要通过该注解标记让自定义的订阅监听器优先。