akka-stream之异常处理

Posted junjiang3

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了akka-stream之异常处理相关的知识,希望对你有一定的参考价值。

背景介绍

在项目中使用了akk-stream的source.queue功能,如下:

Pair<SourceQueueWithComplete<Integer>, Source<Integer, NotUsed>> reqSourceQueue = Source<Integer>queue(1024,OverflowStrategy.backpressure()).preMaterialize(materializer);

但是,如果在使用reqSourceQueue.second().runForeach方法来对流中的元素进行处理时,如果处理代码出现错误,也不抛异常,并且整个流中断了,继续往queue中offer元素也不起作用了。查看官方文档,有以下两种解决方案

recover方法

该方法在流运行的过程中出现异常时,会使用recover中的match进行处理。示例代码如下:

final Materializer mat = ActorMaterializer.create(system);
Source.from(Arrays.asList(0, 1, 2, 3, 4, 5, 6))
    .map(
        n -> 
          if (n < 5) return n.toString();
          else throw new RuntimeException("Boom!");
        )
    .recover(new PFBuilder().match(RuntimeException.class, ex -> "stream truncated").build())
    .runForeach(System.out::println, mat);

但是这种方案也存在一个问题,他在出现错误,并且在recover匹配上以后,会结束整个流,继续往queue中offer不起作用了,显然这不是我们想要的处理方式。

监督机制

和akka actor的监督机制相同,我们同样可以在流运行的过程中使用监督机制,这种监督机制是在流运行时使用的materializer中设置的,具体有以下几种监督机制:

  • Stop 出现错误,立即结束流的运行
  • Resume 出现错误的流元素会被抛弃,流继续正常运行
  • Restart 出现错误的流元素会被抛弃,但是流是会在restart以后继续运行,这也就意味着附加在流上的所有状态都会被清空
    通常Resume方法使我们想要的,因为这样即使一个流元素的处理出现错误,也不会影响整个流的运行,导致用户的请求没有响应等各种情况。
    具体示例代码如下:
final Function<Throwable, Supervision.Directive> decider =
    exc -> 
      //出现该异常时,这个流元素会被抛弃掉
      if (exc instanceof ArithmeticException) return Supervision.resume();
      else return Supervision.stop();
    ;
final Materializer mat =
    ActorMaterializer.create(
        ActorMaterializerSettings.create(system).withSupervisionStrategy(decider), system);
final Source<Integer, NotUsed> source =
    Source.from(Arrays.asList(0, 1, 2, 3, 4, 5)).map(elem -> 100 / elem);
final Sink<Integer, CompletionStage<Integer>> fold = Sink.fold(0, (acc, elem) -> acc + elem);
final CompletionStage<Integer> result = source.runWith(fold, mat);

以上是关于akka-stream之异常处理的主要内容,如果未能解决你的问题,请参考以下文章

重学SpringBoot系列之统一全局异常处理

Python 基础之 异常处理

python之异常处理机制

Python之异常处理模块与包

python之异常处理

python之异常处理