springboot maven 版本 redis stream-CFANZ编程社区

使用 Spring Boot 和 Maven 实现 Redis Stream

引言

在现代分布式系统中，消息队列是实现服务解耦和异步处理的重要组件。Redis Streams 是 Redis 5.0 新增的功能之一，适用于高吞吐量和复杂的消费者场景。本文将介绍如何在 Spring Boot 应用中使用 Maven 集成 Redis Streams，并提供相应的代码示例。

环境准备

首先，你需要有基本的 Java 和 Maven 知识，以及安装好 Redis 服务器。确保你的系统上安装了 JDK 和 Maven，并且能够通过命令行访问它们。

Maven 依赖

在你的 pom.xml 文件中添加以下依赖：

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

这将引入 Spring Boot 对 Redis 的支持，以及 Web 相关的功能。

Redis 配置

在 application.yml 文件中配置 Redis 连接信息：

spring:
  redis:
    host: localhost
    port: 6379

创建 Redis Stream

生产者代码

首先，我们构建一个生产者，向 Redis Stream 发送消息。以下示例创建了一个 MessageProducer 类：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public void sendMessage(String message) {
        stringRedisTemplate.opsForStream().add("mystream", "message", message);
    }
}

消费者代码

接下来，我们创建一个消费者，监听 Redis Stream 中的消息。示例代码如下：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.listener.StreamListener;
import org.springframework.stereotype.Service;

@Service
public class MessageConsumer implements StreamListener {

    @Autowired
    private StreamMessageListenerContainer listenerContainer;

    @Override
    public void onMessage(Message message) {
        System.out.println("Received message: " + message.getBody());
    }

    // 设置监听
    @PostConstruct
    public void init() {
        listenerContainer.start();
    }
}

Stream Listener 容器配置

为了监听 Redis Stream 消息，我们需要配置一个 StreamMessageListenerContainer：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.listener.StreamListener;

@Configuration
public class RedisConfig {

    @Bean
    public StreamMessageListenerContainer messageListenerContainer(RedisMessageListenerContainer container) {
        return StreamMessageListenerContainer.create(container);
    }
}

测试

现在我们可以在 Spring Boot 应用中测试生产者和消费者。以下是一个简单的控制器示例，帮助我们发送消息：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MessageController {
    
    @Autowired
    private MessageProducer messageProducer;

    @PostMapping("/send")
    public void sendMessage(@RequestBody String message) {
        messageProducer.sendMessage(message);
    }
}

启动应用

启动应用后，可以使用 Postman 或 curl 向 /send 接口发送 POST 请求来发送消息。消费者在后台会监听到这些消息。

序列图

下面是一个序列图，表示生产者和消费者之间的交互关系：

sequenceDiagram
    participant Producer
    participant Redis
    participant Consumer

    Producer->>Redis: sendMessage("Hello, Redis Stream!")
    Redis-->>Consumer: deliver message
    Consumer-->>Consumer: process message