問題

用過storm或者jstorm的都知道，如果在bolt代碼中發(fā)生了沒被catch住的異常，所在worker進(jìn)程會退出。本文就從源碼角度分析一下具體設(shè)計(jì)，其實(shí)并不是“有異常然后進(jìn)程崩了”這么簡單。

實(shí)質(zhì)

我們先看BasicBoltExecutor的源碼：

  public void execute(Tuple input) {
    _collector.setContext(input);
    try {
      _bolt.execute(input, _collector);
      _collector.getOutputter().ack(input);
    } catch (FailedException e) {
      if (e instanceof ReportedFailedException) {
        _collector.reportError(e);
      }
      _collector.getOutputter().fail(input);
    }
  }

_bolt.execute(input, _collector) 就是執(zhí)行我們自己編寫的bolt里的excute方法?？梢钥吹剑谶@里，只會catch storm自己定義的FailedException，并且發(fā)送fail消息，標(biāo)記tuple處理失敗， 其余異常則會被放過。

再外層是BoltExecutors的processTupleEvent方法：

    try {
      if (!isSystemBolt && tuple.getSourceStreamId().equals(Common.TOPOLOGY_MASTER_CONTROL_STREAM_ID)) {
        backpressureTrigger.handle(tuple);
      } else {
        bolt.execute(tuple);
      }
    } catch (Throwable e) {
      error = e;
      LOG.error("bolt execute error ", e);
      report_error.report(e);
    }

在這里，所有異常都會被catch住，但是只會進(jìn)行report_error，并不會發(fā)fail消息，相關(guān)tuple只能等超時(shí)才能被標(biāo)記為失敗。

再來看report_error.report(e) 的具體實(shí)現(xiàn)，通過看構(gòu)造函數(shù)，可以看到report_error是一個(gè)TaskReportErrorAndDie類，

  @Override
  public void report(Throwable error) {
    this.reporterror.report(error);
    this.haltfn.run();
  }

在這里，reporterror是一個(gè)AsyncLoopDefaultKill類

  @Override
  public void run() {
    JStormUtils.halt_process(1, "Async loop died!");
  }

這里就是整個(gè)過程的最終步驟了， JStormUtils.halt_process()方法會打印一條"Async loop died!"的日志后將worker進(jìn)程殺死。

思考

通過代碼可以出來，對于jstorm，“異常后worker退出”是一個(gè)故意設(shè)計(jì)出的特性，并非程序不健壯。猜測這一塊的設(shè)計(jì)理念就是對于已知異常，開發(fā)人員自己捕獲并重新拋出FailedException，使相應(yīng)消息失??；未知異常則強(qiáng)制使進(jìn)程直接失敗退出，避免過度的catch導(dǎo)致問題被掩蓋。

不過雖然話是這么說，對這個(gè)設(shè)計(jì)還是持保留意見，畢竟storm和普通的java程序不一樣，storm的worker進(jìn)程在退出后是會自動(dòng)被重啟的，所以這種異常處理方式并不能起到failfast的效果。

相反，worker的持續(xù)重啟，還會帶來一些其他問題。再一個(gè)，不主動(dòng)將消息標(biāo)為失敗，而是等超時(shí)，如果設(shè)置的超時(shí)時(shí)間過長(當(dāng)然超時(shí)時(shí)間太長也不合理)，也會引入一些問題。比如說kafkaSpout, 一條消息沒被ack之前是不會繼續(xù)取后邊的數(shù)據(jù)的，這樣如果有一條數(shù)據(jù)需要等超時(shí)，同分區(qū)下的數(shù)據(jù)在這一個(gè)超時(shí)周期內(nèi)，就都無法被處理了。

從另一方面來說，如果像FailedException一樣處理其他所有異常，由于異常之后可以看到有數(shù)據(jù)fail,也并不會掩蓋問題。

所以說，這一塊的處理邏輯，個(gè)人感覺還是需要斟酌一下。

以上這篇jstorm源碼解析之bolt異常處理方法就是小編分享給大家的全部內(nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評論