如何在 Spring Kafka 中接收来自 KSQL 的流式响应？答案

【问题标题】：How to receive streamed response from KSQL in Spring Kafka?如何在 Spring Kafka 中接收来自 KSQL 的流式响应？
【发布时间】：2019-01-20 20:22:46
【问题描述】：

如何在 java spring boot 应用程序中接收来自 kafka KSQL 服务器的分块响应？

当我对 /query 端点进行休息调用时，我只得到 1 行并且连接关闭。如何保持连接打开并接收多行？

医生说

响应被流回，直到在指定的 LIMIT 语句到达，或客户端关闭连接。

在java中实现这一点的方法是什么？即使是 KTable，我也只能得到 1 行作为回报。

https://docs.confluent.io/current/ksql/docs/developer-guide/api.html#run-a-query-and-stream-back-the-output

【问题讨论】：

标签： java apache-kafka ksqldb

【解决方案1】：

我能够解决的方法如下：

以字符串形式获取响应

逐行解析 JSON 对象（KafkaQueryResponse 是代表 1 行的对象）

    ResponseEntity<String> result = template.exchange("/query",
        HttpMethod.POST,
        new HttpEntity<>(params, headers),
        String.class);

    List<KafkaQueryResponse> array = new ArrayList<>();
    JsonFactory jsonFactory = new JsonFactory();
    try(BufferedReader br = new BufferedReader(new StringReader(result.getBody()))) {
        Iterator<KafkaQueryResponse> value = objectMapper.readValues(jsonFactory.createParser(br), KafkaQueryResponse.class);
        value.forEachRemaining(e -> {
            if (e.getRow() != null) {
                array.add(e);
            }
        });
    }
    array <----  this is the list of JSON objects

KafkaQueryResponse

    @Data
    @JsonIgnoreProperties(ignoreUnknown = true)
    public class KafkaQueryResponse {
        private KafkaQueryRow row;
        private String finalMessage;
        private String errorMessage;

        @Data
        @JsonIgnoreProperties(ignoreUnknown = true)
        public static class KafkaQueryRow {
            private List<Object> columns;
        }
    }

此解决方案不允许以块的形式读取流式响应。它等待整个响应到达客户端，然后关闭连接，然后解析所有 json 对象。

【讨论】：