Spring Boot 项目中发布流式接口支持实时数据向客户端推送

这种方式适用于响应式编程，并且可以充分利用 Reactor 框架的特性实现复杂数据流逻辑。StringUtils.streamFlux的方法。

非ban必选

481人浏览 · 2025-04-17 17:24:39

非ban必选 · 2025-04-17 17:24:39 发布

1、pom依赖添加

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

2、事例代码

package com.pojo.prj.controller;

import com.pojo.common.core.utils.StringUtils;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.Map;

@RestController
public class TestController {

    @GetMapping(value = "/stream/flux", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Map<String,String>> streamFlux() {
        // 每隔 1 秒发送一条数据，共发送 10 条
        String query = "select * from test";
        return Flux.interval(Duration.ofSeconds(1))
                .map(sequence -> StringUtils.streamFlux(query,sequence))
                .take(10);
    }
}

StringUtils.streamFlux的方法

 public static Map<String, String> streamFlux(String query, Long sequence) {
        Map<String, String> map = new HashMap<>();
        map.put(sequence + "", query + " " + sequence);
        return map;
    }

在 @GetMapping 中设置 produces = MediaType.TEXT_EVENT_STREAM_VALUE 表示以 SSE 格式推送数据。
Flux.interval(...) 每隔一秒生成一个递增的数字序列，然后通过 map 操作转换成map消息。
take(10) 限制只发送 10 个数据，流结束后自动关闭。

这种方式适用于响应式编程，并且可以充分利用 Reactor 框架的特性实现复杂数据流逻辑。

测试效果

nginx在配置代理SSE接口时需加一下配置

	        proxy_http_version 1.1;           # 强制使用HTTP/1.1协议‌:
            proxy_buffering off;              # 关闭响应缓冲，确保流式传输‌
            proxy_set_header Connection '';