Python和Java之间相互调用是怎么做到的？ - 51Testing软件测试论坛

public class PythonComd extends BaseComd<PythonConf> {
　　
　　 /**
　　 * 单例，用于和 python 进程通信。
　　 */
　　 @Resource
　　 private PythonServer pythonServer;
　　
　　 /**
　　 * python 进程执行结束后，回调 pythonComd.setReturnResult(dataset);
　　 * execSparkDataset() 方法，返回 this.returnResult;
　　 */
　　 @Setter
　　 private Dataset<Row> returnResult;
　　
　　 /**
　　 * 输出流字符串。
　　 */
　　 private StringBuilder outputStringBuilder = new StringBuilder();
　　
　　 /**
　　 * 错误流字符串。
　　 */
　　 private StringBuilder errorStringBuilder = new StringBuilder();
　　
　　 /**
　　 * 构造方法，传入配置类。
　　 * @param conf
　　 */
　　 public PythonComd(PythonConf conf) {
　　 super(conf);
　　 }
　　
　　
　　 /**
　　 * 以命令行的方式，执行 python 进程。
　　 * @param preDataset 上一条命令生成的结果集对象
　　 * @return
　　 * @throws GplException
　　 */
　　 @Override
　　 public Dataset<Row> execSparkDataset(Dataset<Row> preDataset) throws GplException {
　　 if (preDataset == null) {
　　 throw new GplException("python 命令，未找到结果集！");
　　 }
　　 String script = conf.getExpression();
　　 String fileName = conf.getFilename();
　　 if (StringUtils.isEmpty(script) && StringUtils.isEmpty(fileName)) {
　　 throw new GplException("python 命令，脚本为空！");
　　 }
　　 if (StringUtils.isEmpty(script) && !StringUtils.isEmpty(fileName)) {//如果是执行py文件
　　 File pyFile = new File(NConst.PYTHON_PATH + fileName);
　　 if (!pyFile.exists()) {
　　 throw new GplException("python 命令，GPL 的 python 目录找不到脚本文件：" + fileName);
　　 }
　　 }
　　 if (!StringUtils.isEmpty(AppConfig.PYTHON_ENV)) {
　　 File file = new File(AppConfig.PYTHON_ENV);
　　 if (!file.exists()) {
　　 throw new GplException("python 命令，自定义环境变量，路径不存在：" + AppConfig.PYTHON_ENV);
　　 }
　　 }
　　
　　 String uuid = conf.getUuid();
　　 pythonServer.openSession(uuid, this);
　　 String containerName = uuid;
　　 try {
　　 String scriptPath = null;
　　 if (!StringUtils.isEmpty(script)) {//优先执行脚本，如果没有脚本语句赐执行文件
　　 String md5 = DigestUtils.md5Hex(script);
　　 String tmpName = md5 + ".py";
　　 scriptPath = NConst.PYTHON_PATH + tmpName;
　　 File tmpFile = new File(scriptPath);//临时文件不再结束调用后删除，以后完善成定期清理
　　 FileUtils.write(tmpFile, script, NConst.CHARSET_UTF_8);//把脚本写成临时文件
　　 } else {
　　 scriptPath = NConst.PYTHON_PATH + fileName;
　　 }
　　
　　 List<String> command = new ArrayList<>();
　　 if (!StringUtils.isEmpty(conf.getDocker())) {
　　 // docker run --name containerName -v /home/nyx:/home/nyx --net host python:slim python /home/nyx/test.py 25333 uuid
　　 command.add("docker");
　　 command.add("run");
　　 command.add("--name");
　　 command.add(containerName);//容器名
　　 command.add("-e");
　　 command.add("TZ=Asia/Shanghai");//设置时区
　　 command.add("-v");
　　 command.add(NConst.PYTHON_PATH + ":" + NConst.PYTHON_PATH);// 绑定挂载目录
　　 command.add("--net");
　　 command.add("host");
　　 command.add(conf.getDocker());// docker 镜像名
　　 }
　　
　　 if (StringUtils.isEmpty(AppConfig.PYTHON_ENV)) {
　　 command.add("python");
　　 } else {
　　 command.add(AppConfig.PYTHON_ENV);
　　 }
　　 command.add(scriptPath);//py文件路径
　　 command.add(String.valueOf(pythonServer.getPort()));//端口号
　　 command.add(uuid);
　　 String cmd = String.join(" ", command);
　　 log.info("任务 id：" + conf.getTaskId() + "，python 执行命令：" + cmd);
　　 ProcessBuilder processBuilder = new ProcessBuilder(command);
　　 Process process = processBuilder.start();
　　 this.printProcessStream(process);
　　 int timeout = conf.getTimeout();
　　 log.info("任务 id：" + conf.getTaskId() + "，python 脚本执行超时参数为：" + timeout);
　　 if (timeout > 0) {
　　 boolean isExited = process.waitFor(timeout, TimeUnit.MILLISECONDS);//脚本执行超时退出
　　 if (!isExited) {//如果命令行子进程没有退出，销毁子进程。
　　 process.destroy();
　　 }
　　 } else {
　　 process.waitFor();
　　 }
　　 this.clearTmpFile();//最后执行一下清理目录
　　 } catch (Exception e) {
　　 log.error(e.getMessage(), e);
　　 } finally {
　　 if (!StringUtils.isEmpty(conf.getDocker())) {
　　 try {
　　 Process process = new ProcessBuilder("docker", "rm", "-f", containerName).start();
　　 this.printProcessStream(process);
　　 process.waitFor();
　　 log.info("任务 id：" + conf.getTaskId() + "，docker 执行 python 结束！删除容器：" + containerName);
　　 } catch (Exception e) {
　　 log.error(e.getMessage(), e);
　　 }
　　 }
　　 pythonServer.closeSession(uuid);
　　 }
　　 if (errorStringBuilder.length() > 0) {
　　 String error = errorStringBuilder.toString();
　　 throw new GplException("python 命令，脚本执行异常！", error);
　　 }
　　 if (outputStringBuilder.length() > 0) {
　　 String input = outputStringBuilder.toString();
　　 comdTask.addWarnMsg("python 命令，控制台打印：" + input);
　　 }
　　 if (returnResult == null) {
　　 throw new RuntimeException("python 命令，脚本执行结束，脚本返回结果集为空！");
　　 }
　　 // 此处不要计算 count()，会消耗掉几十秒，直接打印一下字段名就行了。
　　 Map<String, String> fieldMap = new HashMap<>();
　　 StructType structType = returnResult.schema();
　　 StructField[] structFieldArray = structType.fields();
　　 for (StructField structField : structFieldArray) {
　　 fieldMap.put(structField.name(), structField.dataType().simpleString());
　　 }
　　 log.info("任务 id：" + conf.getTaskId() + "，python 命令，返回结果集字段：" + fieldMap.toString());
　　 return this.returnResult;
　　 }
　　
　　 /**
　　 * 清理生成的临时文件
　　 */
　　 private void clearTmpFile() {
　　 File pyPath = new File(NConst.PYTHON_PATH);
　　 File[] fileArray = pyPath.listFiles(f -> {
　　 return f.getName().length() == 35//文件名长度 35 的是临时文件
　　 && System.currentTimeMillis() - f.lastModified() > 86400000;//删除大于 1 天的文件
　　 });
　　 for (File file : fileArray) {
　　 file.delete();
　　 }
　　 }
　　
　　 /**
　　 * 启动两个后台线程，用于接收 python 进程的输出流。
　　 * @param process
　　 */
　　 private void printProcessStream(Process process) {
　　 Thread inputThread = new Thread(() -> {
　　 try (BufferedReader inputReader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
　　 String input = null;
　　 while ((input = inputReader.readLine()) != null) {
　　 outputStringBuilder.append("\n");
　　 outputStringBuilder.append(input);
　　 }
　　 if (outputStringBuilder.length() > 0) {
　　 log.info("任务 id：" + conf.getTaskId() + "，python print(): " + outputStringBuilder.toString());
　　 }
　　 } catch (Exception e) {
　　 log.error(e.getMessage(), e);
　　 }
　　 });
　　 inputThread.setName("gpl-py-input-stream-task-id=" + conf.getTaskId());
　　 inputThread.setDaemon(true);
　　 inputThread.start();
　　
　　 Thread errorThread = new Thread(() -> {
　　 try (BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
　　 String error = null;
　　 while ((error = errorReader.readLine()) != null) {
　　 errorStringBuilder.append("\n");
　　 errorStringBuilder.append(error);
　　 }
　　 if (errorStringBuilder.length() > 0) {
　　 log.error("任务 id：" + conf.getTaskId() + "，python error: " + errorStringBuilder.toString());
　　 }
　　 } catch (Exception e) {
　　 log.error(e.getMessage(), e);
　　 }
　　 });
　　 errorThread.setName("py-error-stream-task-id=" + conf.getTaskId());
　　 errorThread.setDaemon(true);
　　 errorThread.start();
　　 }
　　}

复制代码