storm(3)-任務提交

  • 2019 年 10 月 4 日
  • 筆記

storm job的提交分為本地模式和遠程模式

下面我們先從程式碼入手，分析一下兩者的提交

1.本地模式

2.遠程模式提交

通過上面兩種程式碼的分析發現本地模式和遠程模式還是有著很大的區別

但是如果我們刨根問底會發現其實最終都是一致的

本地模式其實其實使用的是127.0.0.1，如果在storm集群上，藉助storm jar則使用的是storm.yaml中的配置

下面我們講一下通過java的Rumtime exec的方式進行storm jar的提交

public void submitTopologyToMachine(String nimbusAddress, String fileName, List<String> mainArgs, String mainClass, String jarFile){          StringBuffer args = new StringBuffer();          args.append(jarFile).append(" ");          for (String arg: mainArgs){              args.append(arg).append(" ");          }          args.append("-Dstorm.options=nimbus.host=").append(nimbusAddress).append(" ");          args.append(mainClass);          if (fileName != null){              args.append(" ").append(fileName);          }          InputStream is = null;          try {              Runtime rt = Runtime.getRuntime();              String command = "/home/apps/platform/storm/bin/storm/bin/storm jar "+args.toString();              LOG.info("submit topology command is {}", command);              String[] commandStr = {"/bin/sh","-c",command};              Process proc = rt.exec(commandStr);              if (LOG.isDebugEnabled()) {                  is = proc.getErrorStream();                  BufferedReader reader = new BufferedReader(new InputStreamReader(is));                  String line;                  while ((line = reader.readLine()) != null) {                      LOG.debug("submit topology result line {}", line);                  }              }              int exitVal = proc.waitFor();              LOG.info("submit topology result is {}", exitVal==0? "success" : "failure");            } catch (IOException e) {              LOG.error("submit job is exception {}", e);          } catch (InterruptedException e) {              LOG.error("submit job is exception {}", e);          }finally {              try {                  if (is != null){                      is.close();                  }              } catch (IOException e) {                }          }      }

這是利用Runtime.getRuntime().exec()來執行，但是只能在linux上執行，而且還需要提前放置一份storm的安裝包

Tags:

VirMach 便宜 VPS