如何使用Mutiny以响应式方式读取文件行？

你的思路并没有偏离正轨，核心方向是对的——将命令式的逐行读取转化为响应式的Multi<String>流。不过直接使用Multi.createFrom(bReader::readline)存在几个问题：无法自动判断流的终止条件、没有处理IO异常、也不会自动关闭资源。下面是符合Mutiny规范的正确实现方式：

推荐实现：使用资源管理自动关闭流 #

这种方式利用Mutiny的resource()API自动管理BufferedReader的生命周期，无需手动关闭资源，同时处理读取过程中的异常：

import io.smallrye.mutiny.Multi;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

public Multi<String> loadReactive(final InputStream inData) {
    // 1. 定义资源提供者：创建BufferedReader
    // 2. 定义如何从资源生成Multi流
    return Multi.createFrom().resource(
            () -> new BufferedReader(new InputStreamReader(inData, StandardCharsets.UTF_8)),
            reader -> Multi.createFrom().generator(
                    () -> reader, // 初始化状态为BufferedReader
                    (r, emitter) -> {
                        try {
                            String line = r.readLine();
                            if (line != null) {
                                emitter.emit(line); // 发射读取到的行
                            } else {
                                emitter.complete(); // 读取完毕，结束流
                            }
                        } catch (IOException e) {
                            emitter.fail(e); // 传递IO异常给订阅者
                        }
                    }
            )
    );
}

手动管理资源的实现方式 #

如果你需要更细粒度控制资源关闭，也可以手动在流结束或失败时关闭BufferedReader：

import io.smallrye.mutiny.Multi;
import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

public Multi<String> loadReactive(final InputStream inData) {
    return Multi.createFrom().generator(
            () -> new BufferedReader(new InputStreamReader(inData, StandardCharsets.UTF_8)), // 初始化BufferedReader
            (reader, emitter) -> {
                try {
                    String line = reader.readLine();
                    if (line != null) {
                        emitter.emit(line);
                    } else {
                        emitter.complete();
                        reader.close(); // 读取完成后关闭资源
                    }
                } catch (IOException e) {
                    emitter.fail(e);
                    try {
                        reader.close(); // 异常时关闭资源
                    } catch (IOException ex) {
                        // 可根据需求记录日志或忽略
                    }
                }
            }
    );
}

关键说明 #

• 终止条件：通过判断readLine()返回null时调用emitter.complete()，告诉Mutiny流已经结束，避免无限生成。
• 异常处理：捕获IOException并通过emitter.fail()传递给订阅者，符合响应式编程中错误传播的规范。
• 资源管理：推荐使用resource()API，它会在流完成、失败或订阅取消时自动关闭实现AutoCloseable的资源（BufferedReader实现了该接口），避免资源泄漏。

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