java - 如何在java中编写代码来接收text/event-stream事件流并一一打印？

Question

我的需求是通过post发送请求调用API接口(interface)，然后将服务器返回的text/event-stream事件流一一打印出来，不管我使用的是HttpURLConnection、WebFlux还是RestTemplate。客户端等待服务器的完整响应，然后将数据组合成一 block 打印出来。无法实现一件一件退回的效果。有什么解决方案可以满足我的需求吗？这是我通过 crul 发送请求时得到的:

{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "1", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "2", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "3", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "4", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "5", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "6", "index": 0, "logprobs": null, "finish_reason": null}]}
{"id": "cmpl-6olAPoaWSKGOEj9JBlnEipVsKLblW", "object": "text_completion", "created": 1677555729, "choices": [{"text": "7", "index": 0, "logprobs": null, "finish_reason": null}]}

这是我使用 HttpURLConnection 编写的代码:

    HttpURLConnection urlConnection = (HttpURLConnection) new URL(url).openConnection();
    urlConnection.setDoOutput(true);
    urlConnection.setDoInput(true);
    urlConnection.setUseCaches(false);
    urlConnection.setRequestMethod("POST");  
    urlConnection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
    urlConnection.setRequestProperty("Connection", "Keep-Alive");
    urlConnection.setRequestProperty("Charset", "UTF-8");
    urlConnection.setRequestProperty("Authorization", "Bearer " + token);
    urlConnection.setRequestProperty("Accept", "text/event-stream");
    urlConnection.getOutputStream().write(JSON.toJSONString(completionRequest).getBytes(StandardCharsets.UTF_8));

    InputStream inputStream = urlConnection.getInputStream();
    InputStreamReader inputStreamReader = new InputStreamReader(inputStream);
    BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
    String line;
    while ((line = bufferedReader.readLine()) != null) {
        System.out.println("lien:"+line);
    }

这是我使用 webflux 编写的代码:

  HttpHeaders httpHeaders = new HttpHeaders();
  httpHeaders.setContentType(MediaType.parseMediaType("application/json; charset=UTF-8"));
  httpHeaders.setAccept(Arrays.asList(MediaType.APPLICATION_JSON));
  httpHeaders.add("Authorization", "Bearer " + token);
  HttpEntity<String> formEntity = new HttpEntity<String>(completionJSON, httpHeaders);

  Flux.create(emitter -> {
        try {
            List<String> resultList = Arrays.stream(restTemplate.postForObject(url, formEntity, String.class).split("\n")).
                    filter(line -> !line.trim().equals("")).collect(Collectors.toList());
            for (String result : resultList) {
                String message = result.replace("data: ", "");
                if (message.equals("[DONE]")) {
                    return;
                } else {
                    JSONArray choicesJSON = new JSONObject(message).getJSONArray("choices");
                    Object text = new JSONObject(choicesJSON.get(0).toString()).get("text");
                    emitter.next(text);
                }
            }
        } catch (Exception ex) {
            emitter.error(ex);
        }
    }).subscribe(System.out::println);

但是在这两种情况下，输出都是这样的:

{"id": "cmpl-6onbJlmNNZBShLvHeJhdOMRAhR2x6", "object": "text_completion", "created": 1677565085, "choices": [{"text":1 2 3 4 5 6 7", "index": 0, "logprobs": null, "finish_reason": null}]}

很明显结果已经合并了，而不是逐条返回打印，请问有没有办法让客户端逐条接收数据并打印？

Answer 1

假设服务器使用 text/event-stream 媒体类型返回数据，您可以使用 WebClient 使用流，然后作为 Flux 发出。

@GetMapping(path = "/client", produces = TEXT_EVENT_STREAM_VALUE)
Flux<String> client() {
        return webclient.post()
                .uri("/server")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .header(HttpHeaders.CONNECTION, "keep-alive")
                .retrieve()
                .bodyToFlux(String.class);
}

您可以通过模拟来自同一 Controller 的服务器事件来测试它

@PostMapping(path = "/server", produces = TEXT_EVENT_STREAM_VALUE)
Flux<String> server() {
    return Flux.interval(Duration.ofSeconds(1))
         .map(i -> "value-" + i);
}

这里是一个curl命令来验证

curl -v -H "Accept: text/event-stream" http://localhost:8080/client

*   Trying 127.0.0.1:8080...
* Connected to localhost (127.0.0.1) port 8080 (#0)
> GET /client HTTP/1.1
> Host: localhost:8080
> User-Agent: curl/7.85.0
> Accept: text/event-stream
> 
< HTTP/1.1 200 OK
< transfer-encoding: chunked
< Content-Type: text/event-stream;charset=UTF-8
< 
data:value-0

data:value-1

data:value-2