下面的代码可以工作，但是这是完全同步的代码（可能只有预取功能）。它使用Apache Http Components。

HttpClientBuilder.create().build().use { client ->
    val responseHandler = ResponseHandler { response ->
        response.entity.content.use { content ->
            content.bufferedReader().use { buffered ->
                // create class, which can process each line. 
                val processor = StreamedLinesProcessor<TResult>()

                do {
                    val nextLine = buffered.readLine()
                    val needContinue = processor.processNextLine(nextLine)
                } while (needContinue)
                processor.getResult()
            }
        }
    }

    client.execute(HttpGet(url.toString()), responseHandler)
}

我还没有找到一个嵌入式的，我正在使用类似下面代码的东西。我假设你正在使用单字符\n行结束，并且\n在这个类看到的任何编码中总是被编码为单字节字符，例如UTF16将不被支持。

Mono<ResponseEntity<Flux<String>>>  = webclient
                .post()
                .uri(...)
                .contentType(...)
                .bodyValue(...)
                .retrieve()
                .toEntityFlux(new LineBodyExtractor());

import lombok.NonNull;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.util.MimeType;
import org.springframework.web.reactive.function.BodyExtractor;
import reactor.core.publisher.Flux;

import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Optional.of;

/**
 * Only supports single char separators and charsets
 **/
public class LineBodyExtractor implements BodyExtractor<Flux<String>, ClientHttpResponse> {
    private static final String SEPARATOR_CHAR = "\n";

    @NonNull
    @Override
    public Flux<String> extract(ClientHttpResponse inputMessage, @NonNull Context ignored) {
        var charSet = of(inputMessage.getHeaders())
                .map(HttpHeaders::getContentType)
                .map(MimeType::getCharset)
                .orElse(UTF_8);
        var separatorBytes = SEPARATOR_CHAR.getBytes(charSet);
        if (separatorBytes.length != 1) {
            throw new IllegalStateException("Charset %s doesn't encode separator %s to a single byte and is unsupported".formatted(charSet, SEPARATOR_CHAR));
        }
        byte separator = separatorBytes[0];
        return inputMessage.getBody()
                .flatMap(buf -> this.splitBy(separator, buf))
                .windowUntil(buf -> buf.indexOf(i -> i == separator, 0) == 0)
                // need to join byte buffers before calling to string, UTF8 is multi byte
                .flatMap(DataBufferUtils::join)
                .map(buf -> buf.toString(charSet));
    }

    Flux<DataBuffer> splitBy(Byte separator, DataBuffer buf) {
        if (buf.capacity() <= 0) { // e.g. an empty trailing line should still emit an event
            return Flux.just(buf);
        }
        Flux<DataBuffer> r = Flux.empty();
        int separatorIndex;
        while ((separatorIndex = buf.indexOf(i -> i == separator, 0)) >= 0) {
            r = Flux.concat(r, Flux.just(buf.split(separatorIndex), buf.split(1)));
        }
        return Flux.concat(r, Flux.just(buf)).filter(b -> b.capacity() > 0);
    }
}