实现高效消息处理:在 Spring Boot 中使用 Redis Stream

2023-12-14 14:33:28

Redis,作为一个多功能的内存数据结构存储,提供了各种数据结构来满足不同的需求。Redis Stream,作为其提供的一种数据结构,是构建复杂消息处理系统的理想选择。与 Redis 的传统发布订阅模式相比,Stream 提供了更高级的特性,如消费者组和消息确认机制。

理解 Redis Stream
Redis Stream 是 Redis 5.0 引入的一个新的数据类型,它本质上是一个日志数据结构。每个条目都包含一个唯一的序列号和一个键值对。Stream 主要用于存储一系列的事件或消息,这些消息可以由一个或多个消费者消费。

特性
消费者组:类似于 Kafka 中的消费者组,可以确保每条消息只被组内的一个消费者处理。
消息确认:消费者在处理完消息后可以发送确认,确保消息不会被再次处理。
持久化:Stream 中的消息是持久化的,即使在服务器重启后也不会丢失。
应用场景
假设我们有一个应用管理系统,需要在添加新的应用时通知其他服务进行处理。如果用redis发布订阅机制来写的话,若消费者服务多节点,又会发生重复消费的问题,而redis发布订阅机制事实上并没有消费者组的概率,无法避免这个问题。这个过程非常适合使用 Redis Stream 来实现,stream中有消费者组的概念,和rocketmq一样,我们可以创建一个流来保存新应用的信息,并设置消费者来处理这些信息。

Spring Boot 中的实现
配置 RedisTemplate
首先,确保在你的 Spring Boot 应用中已经配置了 RedisTemplate。以下是一个基础的配置示例:

 
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        return template;
    }
}

发布消息
在我们的应用管理服务中,当添加一个新应用时,我们需要将应用的信息发送到 Redis Stream。

@RestController
@RequestMapping("/api/applications")
public class ApplicationController {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @PostMapping("/add")
    public ResponseEntity<String> addApplication(@RequestBody SysApplication sysApplication) {
        // 保存应用逻辑...
        
        // 发布消息到 Redis Stream
        Map<String, Object> messageData = new HashMap<>();
        messageData.put("app", sysApplication);
        redisTemplate.opsForStream().add("applications_stream", messageData);

        return ResponseEntity.ok("Application added successfully");
    }
}

消费消息
创建一个服务来定期从 Stream 读取并处理消息。并设置消费者组,这样就避免了多节点情况下,重复消费的问题

@Service
public class ApplicationStreamConsumer {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Scheduled(fixedRate = 10000)
    public void consumeMessages() {
        String streamKey = "applications_stream";
        String groupName = "app_group";
        String consumerName = "app_consumer_" + UUID.randomUUID().toString();

        // 创建消费者组
        try {
            redisTemplate.opsForStream().createGroup(streamKey, groupName);
        } catch (Exception e) {
            // 组可能已经存在,忽略异常
        }

        // 读取并处理消息
        List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream()
            .read(Consumer.from(groupName, consumerName),
                  StreamReadOptions.empty().block(Duration.ofMillis(10000)),
                  StreamOffset.create(streamKey, ReadOffset.lastConsumed()));

        for (MapRecord<String, Object, Object> message : messages) {
            // 处理消息逻辑...
            // 确认消息
            redisTemplate.opsForStream().acknowledge(streamKey, groupName, message.getId());
        }
    }
}

在这个服务中,我们使用 @Scheduled 注解定期执行消息消费逻辑。每次从 Stream 读取消息后,我们处理消息并发送确认,以确保消息不会被重复处理。

使用 Redis Stream 在 Spring Boot 中实现消息的发布和消费是一个高效且可靠的方法,特别适用于构建需要确保消息顺序和可靠性的复杂系统。通过利用 Redis Stream 的高级特性,如消费者组和消息确认,我们可以构建一个强大的事件驱动架构,为微服务和分布式系统提供强有力的支持。

文章来源:https://blog.csdn.net/weixin_43866043/article/details/134863673
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。