如何从Akka Streams Sink抛出的异常中恢复？
如何从Akka Streams Sink抛出的异常中恢复？
提问于 2020-03-16 22:31:36
如何从Akka Streams的Sink中抛出的异常中恢复？
简单的例子：
Source<Integer, NotUsed> integerSource = Source.from(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9));
integerSource.runWith(Sink.foreach(x -> {
if (x == 4) {
throw new Exception("Error Occurred");
}
System.out.println("Sink: " + x);
}), system);输出：
Sink: 1
Sink: 2
Sink: 3如何处理异常并从源代码转到下一个元素？(又名5,6,7,8,9)
回答 1
Stack Overflow用户
回答已采纳
发布于 2020-03-16 23:01:21
默认情况下，当抛出异常时，supervision strategy会停止流。要更改监督策略以丢弃导致异常的消息并继续处理下一条消息，请使用"resume“策略。例如：
final Function<Throwable, Supervision.Directive> decider =
exc -> {
return Supervision.resume();
};
final Sink<Integer, CompletionStage<Done>> printSink =
Sink.foreach(x -> {
if (x == 4) {
throw new Exception("Error Occurred");
}
System.out.println("Sink: " + x);
});
final RunnableGraph<CompletionStage<Done>> runnableGraph =
integerSource.toMat(printSink, Keep.right());
final RunnableGraph<CompletionStage<Done>> withResumingSupervision =
runnableGraph.withAttributes(ActorAttributes.withSupervisionStrategy(decider));
final CompletionStage<Done> result = withResumingSupervision.run(system);您还可以为不同类型的异常定义不同的监督策略：
final Function<Throwable, Supervision.Directive> decider =
exc -> {
if (exc instanceof MySpecificException) return Supervision.resume();
else return Supervision.stop();
};页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/60707839
复制相关文章
相似问题
腾讯云开发者
