Spring Kafka与Spring Boot中如何接收KSQL的流式及分块响应？

我之前也踩过这个坑——用普通的RestTemplate调用KSQL的/query端点，只能拿到一行数据就断开连接，完全没实现官方文档说的“流式返回直到LIMIT或客户端关闭”。问题的核心是：KSQL的/query接口用的是分块传输编码（Chunked Transfer Encoding），而传统阻塞式客户端不会持续监听流，拿到第一块数据就认为响应完成了。下面是两种经过验证的实现方式：

一、推荐方案：用Spring WebClient处理流式响应 #

WebClient是Spring 5+推出的非阻塞式HTTP客户端，天生支持流式处理分块响应，非常适合这种场景。

1. 先配置WebClient Bean #

@Configuration
public class KsqlClientConfig {
    @Bean
    public WebClient ksqlWebClient() {
        return WebClient.builder()
                .baseUrl("http://your-ksql-server-host:8088") // 替换成你的KSQL服务器地址
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build();
    }
}

2. 发起流式查询并处理每行数据 #

@Service
public class KsqlStreamService {
    private final WebClient ksqlWebClient;
    private final ObjectMapper objectMapper;

    @Autowired
    public KsqlStreamService(WebClient ksqlWebClient, ObjectMapper objectMapper) {
        this.ksqlWebClient = ksqlWebClient;
        this.objectMapper = objectMapper;
    }

    public void subscribeToKsqlStream() {
        // 注意：不管是Stream还是KTable，要持续接收更新必须加EMIT CHANGES
        String ksqlQuery = "SELECT * FROM your_topic_stream EMIT CHANGES;";
        // 如果是KTable，比如要监听状态变化：SELECT * FROM your_ktable EMIT CHANGES;

        ksqlWebClient.post()
                .uri("/query")
                .bodyValue(Map.of(
                        "ksql", ksqlQuery,
                        "streamsProperties", Map.of("auto.offset.reset", "earliest")
                ))
                .retrieve()
                .bodyToFlux(String.class) // 按分块读取每一行JSON数据
                .subscribe(
                        // 处理每行结果
                        line -> {
                            try {
                                // 把JSON行转成你自己的POJO
                                YourDataModel data = objectMapper.readValue(line, YourDataModel.class);
                                System.out.println("Received updated data: " + data);
                                // 这里可以做业务处理：比如存入数据库、触发事件等
                            } catch (JsonProcessingException e) {
                                e.printStackTrace();
                            }
                        },
                        // 处理错误（比如KSQL服务器断开、查询语法错误）
                        error -> {
                            System.err.println("Stream error occurred: " + error.getMessage());
                            // 可以在这里加重连逻辑
                        },
                        // 流结束时的操作（比如达到LIMIT、主动关闭连接）
                        () -> System.out.println("KSQL stream completed")
                );
    }
}

二、备选方案：用JDK原生HttpClient（JDK 11+） #

如果你的项目没用到WebFlux，也可以用JDK 11自带的HttpClient来实现，不需要额外依赖：

public void streamWithNativeHttpClient() throws IOException, InterruptedException {
    HttpClient client = HttpClient.newHttpClient();
    String ksqlQuery = "SELECT * FROM your_ktable EMIT CHANGES LIMIT 5;";

    HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create("http://your-ksql-server-host:8088/query"))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(
                    "{\"ksql\":\"" + ksqlQuery + "\"," +
                    "\"streamsProperties\":{\"auto.offset.reset\":\"earliest\"}}"
            ))
            .build();

    // 按行读取分块响应
    client.sendAsync(request, HttpResponse.BodyHandlers.ofLines())
            .thenApply(HttpResponse::body)
            .thenAccept(lines -> lines.forEach(line -> {
                System.out.println("Received KSQL chunk: " + line);
                // 同样可以在这里解析JSON做业务处理
            }))
            .join();
}

三、关键注意事项 #

1. 必须加EMIT CHANGES子句：这是很多人踩坑的点！如果查询KTable时不加这个，KSQL只会返回当前的状态快照（可能是一行或多行，但不会持续推送更新），然后直接关闭连接。只有加了EMIT CHANGES，才会保持连接并持续推送新的状态变化。
2. 避免用RestTemplate：RestTemplate是阻塞式客户端，它会等待整个响应完成后才返回，而KSQL的流式响应永远不会“完成”（除非达到LIMIT或断开），所以用它只能拿到第一块数据就卡住或断开。
3. 连接保持与重连：KSQL会保持连接直到以下情况：
   • 查询语句中的LIMIT条件满足；
   • 客户端主动取消订阅/关闭连接；
   • KSQL服务器出现故障或重启。
     所以如果需要长期监听，建议在代码中加入重连逻辑，处理连接断开的情况。

四、先验证KSQL端是否正常 #

可以先用curl测试KSQL的流式响应是否正常，排除服务器端的问题：

curl -X POST "http://your-ksql-server-host:8088/query" \
     -H "Content-Type: application/json" \
     -d '{"ksql":"SELECT * FROM your_stream EMIT CHANGES;","streamsProperties":{"auto.offset.reset":"earliest"}}'

如果curl能持续输出多行JSON，说明KSQL配置没问题，再排查Java代码的逻辑。

内容的提问来源于stack exchange，提问作者Sergei Ledvanov