快捷導(dǎo)航

jstorm源碼解析之bolt異常處理方法

更新時(shí)間：2017年08月04日 09:03:03 投稿：jingxian

下面小編就為大家?guī)?lái)一篇jstorm源碼解析之bolt異常處理方法。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

問(wèn)題

用過(guò)storm或者jstorm的都知道，如果在bolt代碼中發(fā)生了沒(méi)被catch住的異常，所在worker進(jìn)程會(huì)退出。本文就從源碼角度分析一下具體設(shè)計(jì)，其實(shí)并不是“有異常然后進(jìn)程崩了”這么簡(jiǎ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方法?？梢钥吹?，在這里，只會(huì)catch storm自己定義的FailedException，并且發(fā)送fail消息，標(biāo)記tuple處理失敗，其余異常則會(huì)被放過(guò)。

再外層是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);
    }

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

再來(lái)看report_error.report(e) 的具體實(shí)現(xiàn)，通過(guò)看構(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è)過(guò)程的最終步驟了， JStormUtils.halt_process()方法會(huì)打印一條"Async loop died!"的日志后將worker進(jìn)程殺死。

思考

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

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

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

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

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

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

您可能感興趣的文章: