java - 如何从 Akka Streams Sink 中抛出的异常中恢复?
问题描述
如何从 Akka Streams 的接收器中抛出的异常中恢复?
简单示例:
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)
解决方案
默认情况下,监督策略会在抛出异常时停止流。要更改监督策略以删除导致异常的消息并继续处理下一条消息,请使用“恢复”策略。例如:
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();
};
推荐阅读
- bash - 自动创建证书 OpenVPN
- javascript - Typescript 中的“HTMLElement”类型不存在属性“内容”
- quarkus - Quarkus 应用程序启动 bean 初始化两次
- github - 我是否拥有私有 Github 存储库中的代码?
- oracle - 如何在 oracle 查询中使用 into 子句为变量分配多个值?
- python - 用for循环python求解方程
- selenium - 如何通过 Selenium 使用 css-selector 查找元素
- php - VS Code PHP Intelephense 扩展是否有配置标志来关闭参数提示?
- python - 使用 Pyspark 从关系数据集构建层次结构
- python - 创建实例时默认调用 __init__