akka-stream之异常处理

叁歲伎倆 2023-10-10 09:59 69阅读 0赞

### 背景介绍 ###

在项目中使用了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);

转载于:https://www.cnblogs.com/junjiang3/p/11473789.html