标签:truncate 方法 rem cto 处理 没有 ret eof UNC
在项目中使用了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中的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中设置的,具体有以下几种监督机制:
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);
标签:truncate 方法 rem cto 处理 没有 ret eof UNC
原文地址:https://www.cnblogs.com/junjiang3/p/11473789.html