Redis Stream轻松实现SpringBoot消息队列！超实用攻略在此！

后端

2023-08-24 18:12:46

SpringBoot与Redis Stream携手共进，打造高效消息队列系统

什么是Redis Stream？

Redis Stream是一个流处理数据结构，以其超高性能、可靠性和可扩展性而闻名。通过Redis Stream，你可以以发布/订阅（Pub/Sub）的方式发送和接收数据，轻松构建消息队列系统。

为何选择SpringBoot和Redis Stream？

SpringBoot是一个轻量级Java开发框架，以其简单易用而备受欢迎。而Redis Stream作为Redis 5.0的杀手锏，正是构建消息队列系统的绝佳选择。将两者结合，你将获得：

极致性能： Redis Stream的高处理速度让你轻松应对海量数据处理。
坚如磐石的可靠性： Redis Stream支持持久化，确保数据万无一失。
随需应变的可扩展性： Redis Stream可根据需求轻松扩展，满足业务不断增长。

SpringBoot+Redis Stream实战

步骤1：整合Redis Stream

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.core.ReactiveRedisOperations;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@SpringBootApplication
public class RedisStreamApplication {

    public static void main(String[] args) {
        SpringApplication.run(RedisStreamApplication.class, args);
    }

    // 配置Redis连接工厂
    @Bean
    public ReactiveRedisConnectionFactory connectionFactory() {
        return new LettuceConnectionFactory();
    }

    // 配置Redis操作对象
    @Bean
    public ReactiveRedisOperations<String, String> redisOperations(ReactiveRedisConnectionFactory connectionFactory) {
        ReactiveRedisTemplate<String, String> template = new ReactiveRedisTemplate<>();
        template.setConnectionFactory(connectionFactory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        return template;
    }
}

步骤2：发送消息

import org.springframework.data.redis.core.ReactiveRedisOperations;
import reactor.core.publisher.Flux;

@Service
public class MessageService {

    private final ReactiveRedisOperations<String, String> redisOperations;

    public MessageService(ReactiveRedisOperations<String, String> redisOperations) {
        this.redisOperations = redisOperations;
    }

    public Flux<Long> sendMessage(String streamKey, String message) {
        return redisOperations.opsForStream()
                .add(streamKey, message)
                .flatMapMany(Flux::just);
    }
}

步骤3：接收消息

import org.springframework.data.redis.core.ReactiveRedisOperations;
import reactor.core.publisher.Flux;

@Service
public class MessageConsumerService {

    private final ReactiveRedisOperations<String, String> redisOperations;

    public MessageConsumerService(ReactiveRedisOperations<String, String> redisOperations) {
        this.redisOperations = redisOperations;
    }

    public Flux<String> receiveMessage(String streamKey, String consumerGroup) {
        return redisOperations.opsForStream()
                .createGroup(streamKey, consumerGroup)
                .thenMany(redisOperations.opsForStream()
                        .readGroup(streamKey, consumerGroup, 0, 100))
                .flatMap(message -> {
                    // 处理消息
                    return redisOperations.opsForStream()
                            .acknowledge(streamKey, consumerGroup, message.getId());
                });
    }
}