本文介绍如何使用 completablefuture 等异步机制，避免轮询（ready() + sleep）实现多个 inputstream 的真正并发阻塞读取，提升 i/o 效率与响应性。
本文介绍如何使用 completablefuture 等异步机制，避免轮询（ready() + sleep）实现多个 inputstream 的真正并发阻塞读取，提升 i/o 效率与响应性。
在 Java 标准 I/O 模型中，InputStream 本身不支持类似 Unix select() 或 NIO Selector 的多路复用阻塞等待——尤其当输入源是传统阻塞式流（如 Process.getInputStream()、Socket 输入流等）时，无法通过单次调用“监听多个流并返回就绪者”。你当前使用的 ready() 轮询 + sleep() 方案不仅消耗 CPU（即使休眠仍属忙等变体），还会引入延迟，且无法保证实时性。
真正的解决方案是将每个流的阻塞读取委托给独立线程，并通过异步协调机制聚合结果。CompletableFuture.anyOf() 正是为此类场景设计的理想工具：它能阻塞等待任意一个 CompletableFuture 完成，并立即返回其结果，从而模拟出“select(p1, p2)”语义。
以下是一个完整、可运行的示例，封装了从两个 BufferedReader 中非轮询地读取首行：
import java.io.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class MultiInputStreamReader {
public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
// 示例：假设 k 和 k2 是已启动的 Process 实例（如 Runtime.exec）
// BufferedReader p = new BufferedReader(new InputStreamReader(k.getInputStream()));
// BufferedReader p2 = new BufferedReader(new InputStreamReader(k2.getInputStream()));
// 为演示，我们使用带延迟的内存流模拟阻塞输入
BufferedReader p = new BufferedReader(new StringReader("Hello from stream 1\n"));
BufferedReader p2 = new BufferedReader(new StringReader("Hi from stream 2\n"));
// 启动两个异步读取任务：各自阻塞直到读到一行
CompletableFuture<string> future1 = CompletableFuture.supplyAsync(() -> {
try {
String line = p.readLine();
return line != null ? "[Stream-1] " + line : null;
} catch (IOException e) {
throw new UncheckedIOException(e);
}
});
CompletableFuture<string> future2 = CompletableFuture.supplyAsync(() -> {
try {
String line = p2.readLine();
return line != null ? "[Stream-2] " + line : null;
} catch (IOException e) {
throw new UncheckedIOException(e);
}
});
// 阻塞等待任一完成（等效于 select）
Object result = CompletableFuture.anyOf(future1, future2).get();
System.out.println("First available input: " + result);
// 注意：anyOf 返回 Object，需安全转型；更健壮做法是使用 thenAccept/thenCompose 组合
// 若需持续监听（如循环读取多行），应将上述逻辑封装进递归或循环结构，并注意资源关闭
}
}</string></string>
⚠️ 关键注意事项：
- CompletableFuture.anyOf() 返回的是 Object，需手动类型转换或改用 thenAccept() 链式处理以避免 ClassCastException；
- 每个 supplyAsync 默认使用 ForkJoinPool.commonPool()，若流读取可能长时间阻塞（如网络延迟），建议显式传入自定义线程池（如 Executors.newCachedThreadPool()），防止耗尽公共池资源；
- 此方案适用于首次就绪读取；如需持续监听多行，应在每个 CompletableFuture 完成后立即启动下一轮读取任务，形成流水线；
- 若底层流支持 NIO（如 SocketChannel），更高效的方式是迁移到 java.nio.channels.Selector + AsynchronousFileChannel，但需重构为通道模型，不适用于传统 InputStream 包装场景。
综上，虽然 Java 没有原生 select(InputStream...)，但借助 CompletableFuture.anyOf() 结合异步执行，即可优雅、简洁、无轮询地实现多流竞争式阻塞读取，兼顾可读性与工程实用性。
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
