|
|
@@ -0,0 +1,77 @@
|
|
|
+package space.anyi.process;
|
|
|
+
|
|
|
+import java.io.BufferedReader;
|
|
|
+import java.io.File;
|
|
|
+import java.io.IOException;
|
|
|
+import java.io.InputStreamReader;
|
|
|
+import java.io.OutputStream;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.nio.file.Paths;
|
|
|
+import java.util.List;
|
|
|
+import java.util.concurrent.CompletableFuture;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 示例五:综合运用——把子进程当作一条"处理管道"调用外部工具。
|
|
|
+ *
|
|
|
+ * 案例点:runPipeline()
|
|
|
+ * 完整走一遍生产环境调用外部工具的典型流程:
|
|
|
+ * 1. 启动子进程(子进程为 src/main/resources/code/io/Main.java:逐行回显 stdin、EOF 后打印 stderr 统计);
|
|
|
+ * 2. 异步采集子进程 stdout 为结果列表、异步消费 stderr;
|
|
|
+ * 3. 向子进程 stdin 喂入数据;
|
|
|
+ * 4. 等待退出码并取回处理结果、校验回显行数。
|
|
|
+ */
|
|
|
+public class ProcessPipelineExample {
|
|
|
+ private static final Logger log = LoggerFactory.getLogger(ProcessPipelineExample.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);
|
|
|
+ runPipeline();
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 案例点:管道式调用外部工具。喂数据进去、异步取回输出、拿到退出码、校验结果 */
|
|
|
+ private static void runPipeline() throws Exception {
|
|
|
+ // 1. 启动子进程(外部工具)
|
|
|
+ Process process = new ProcessBuilder(Jdk.java(), "-Dfile.encoding=UTF-8", "Main")
|
|
|
+ .directory(WORK_DIR)
|
|
|
+ .start();
|
|
|
+
|
|
|
+ // 2. 异步采集 stdout 成 List<String>,并异步消费 stderr(避免管道写满阻塞子进程)
|
|
|
+ CompletableFuture<List<String>> resultFuture = CompletableFuture.supplyAsync(() ->
|
|
|
+ new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))
|
|
|
+ .lines()
|
|
|
+ .collect(Collectors.toList()));
|
|
|
+ CompletableFuture<Void> stderrDrain = CompletableFuture.runAsync(() ->
|
|
|
+ new BufferedReader(new InputStreamReader(process.getErrorStream(), StandardCharsets.UTF_8))
|
|
|
+ .lines()
|
|
|
+ .forEach(line -> log.info("[stderr] {}", line)));
|
|
|
+
|
|
|
+ // 3. 向子进程 stdin 喂入 3 行待处理数据,关闭输入流表示 EOF
|
|
|
+ try (OutputStream stdin = process.getOutputStream()) {
|
|
|
+ for (int i = 1; i <= 3; i++) {
|
|
|
+ writeLine(stdin, "待处理数据-" + i);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ // 4. 等待退出码,并取回异步收集的处理结果
|
|
|
+ int exit = process.waitFor();
|
|
|
+ List<String> result = resultFuture.get();
|
|
|
+ stderrDrain.get();
|
|
|
+ log.info("退出码: {}", exit);
|
|
|
+ log.info("回显结果: {}", result);
|
|
|
+
|
|
|
+ // 5. 校验:回显行数应等于喂入行数(3 行)
|
|
|
+ long echoed = result.stream().filter(line -> line.startsWith("child: echo")).count();
|
|
|
+ log.info("回显行数: {},符合预期: {}", echoed, echoed == 3);
|
|
|
+ }
|
|
|
+
|
|
|
+ private static void writeLine(OutputStream out, String line) throws IOException {
|
|
|
+ out.write(line.getBytes(StandardCharsets.UTF_8));
|
|
|
+ out.write("\r\n".getBytes(StandardCharsets.UTF_8));
|
|
|
+ }
|
|
|
+}
|