|
|
@@ -0,0 +1,106 @@
|
|
|
+package space.anyi.process;
|
|
|
+
|
|
|
+import java.io.BufferedReader;
|
|
|
+import java.io.File;
|
|
|
+import java.io.IOException;
|
|
|
+import java.io.InputStream;
|
|
|
+import java.io.InputStreamReader;
|
|
|
+import java.io.OutputStream;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.nio.file.Paths;
|
|
|
+import java.util.concurrent.CompletableFuture;
|
|
|
+
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 示例二补充:异步处理子进程的输入和输出。
|
|
|
+ *
|
|
|
+ * 两个案例点各用一个 method 演示:
|
|
|
+ * <ul>
|
|
|
+ * <li>asyncByThread() —— 多线程异步读取 stdout / stderr</li>
|
|
|
+ * <li>asyncByCompletableFuture() —— CompletableFuture 异步读取输出,并结合 onExit() 在进程结束回调</li>
|
|
|
+ * </ul>
|
|
|
+ * 子进程为 src/main/resources/code/io/Main.java。
|
|
|
+ */
|
|
|
+public class ProcessAsyncExample {
|
|
|
+ private static final Logger log = LoggerFactory.getLogger(ProcessAsyncExample.class);
|
|
|
+ private static final File WORK_DIR = Paths.get(System.getProperty("user.dir"),
|
|
|
+ "src", "main", "resources", "code", "io").toFile();
|
|
|
+
|
|
|
+ public static void main(String[] args) throws Exception {
|
|
|
+ Jdk.compileFixture(WORK_DIR);
|
|
|
+ asyncByThread();
|
|
|
+ asyncByCompletableFuture();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 案例点:多线程异步读取输出。
|
|
|
+ * 每个流一个独立线程读取,主线程不阻塞在读取上;
|
|
|
+ * 同时避免子进程输出超过管道缓冲区(约 64KB)时写阻塞、永不退出。
|
|
|
+ */
|
|
|
+ private static void asyncByThread() throws Exception {
|
|
|
+ log.info("== 多线程异步读取 ==");
|
|
|
+ Process process = startChild();
|
|
|
+ Thread stdoutDrain = drain("thread-stdout", process.getInputStream());
|
|
|
+ Thread stderrDrain = drain("thread-stderr", process.getErrorStream());
|
|
|
+
|
|
|
+ // 写入一行输入,子进程回显到 stdout,读到 EOF 后打印 stderr 统计
|
|
|
+ try (OutputStream stdin = process.getOutputStream()) {
|
|
|
+ stdin.write("hello thread\r\n".getBytes(StandardCharsets.UTF_8));
|
|
|
+ }
|
|
|
+ // 等待两个读取线程读到 EOF(即子进程退出)
|
|
|
+ stdoutDrain.join();
|
|
|
+ stderrDrain.join();
|
|
|
+ process.waitFor();
|
|
|
+ log.info("退出码: {}", process.exitValue());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 案例点:CompletableFuture 异步读写。
|
|
|
+ * stdout / stderr 的读取各封装成一个 CompletableFuture 任务;
|
|
|
+ * process.onExit() 在子进程结束时自动完成,可注册回调打印退出码。
|
|
|
+ */
|
|
|
+ private static void asyncByCompletableFuture() throws Exception {
|
|
|
+ log.info("== CompletableFuture 异步读取 ==");
|
|
|
+ Process process = startChild();
|
|
|
+
|
|
|
+ // stdout 读取任务
|
|
|
+ CompletableFuture<Void> stdoutFuture = CompletableFuture.runAsync(() ->
|
|
|
+ new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))
|
|
|
+ .lines()
|
|
|
+ .forEach(line -> log.info("[cf-stdout] {}", line)));
|
|
|
+ // stderr 读取任务
|
|
|
+ CompletableFuture<Void> stderrFuture = CompletableFuture.runAsync(() ->
|
|
|
+ new BufferedReader(new InputStreamReader(process.getErrorStream(), StandardCharsets.UTF_8))
|
|
|
+ .lines()
|
|
|
+ .forEach(line -> log.info("[cf-stderr] {}", line)));
|
|
|
+ // 进程结束时 onExit() 完成,注册回调打印退出码
|
|
|
+ CompletableFuture<Void> exitFuture = process.onExit()
|
|
|
+ .thenAccept(p -> log.info("onExit 回调: 退出码 {}", p.exitValue()));
|
|
|
+
|
|
|
+ // 写入一行输入,然后等待所有异步任务结束
|
|
|
+ try (OutputStream stdin = process.getOutputStream()) {
|
|
|
+ stdin.write("hello future\r\n".getBytes(StandardCharsets.UTF_8));
|
|
|
+ }
|
|
|
+ CompletableFuture.allOf(stdoutFuture, stderrFuture, exitFuture).join();
|
|
|
+ }
|
|
|
+
|
|
|
+ private static Process startChild() throws IOException {
|
|
|
+ return new ProcessBuilder(Jdk.java(), "-Dfile.encoding=UTF-8", "Main")
|
|
|
+ .directory(WORK_DIR)
|
|
|
+ .start();
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 起一个后台线程把给定输入流按行读取并打日志,读到 EOF(子进程退出)后结束 */
|
|
|
+ private static Thread drain(String tag, InputStream in) {
|
|
|
+ Thread thread = new Thread(
|
|
|
+ () -> new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))
|
|
|
+ .lines()
|
|
|
+ .forEach(line -> log.info("[{}] {}", tag, line)),
|
|
|
+ "drain-" + tag);
|
|
|
+ thread.setDaemon(true);
|
|
|
+ thread.start();
|
|
|
+ return thread;
|
|
|
+ }
|
|
|
+}
|