项目反应器的助焊剂在处理错误期间获取失败项目

Question

我需要处理Flux流错误，为此我需要知道到底是什么项目失败了。似乎方法doOnError应该适合处理错误，但有了它，我只能得到异常，而不是失败的项目。有没有办法同时获得失败的项目和异常？

private void testFluxIterableFlow() {
Flux.fromIterable(Arrays.asList(1, 2, 3, 4, 5))
.map(this::process)
.doOnError(ex -> {
...
})
.doOnNext(processedValue -> ...)
.subscribe();
}
private String process(Integer value) {
if (value == 4) {
throw new RuntimeException("error...");
}
return "processed " + value;
}

在此示例中，我需要在错误处理程序中接收失败的项目4和异常消息。

Answer 1

这里有点晚了，但以防其他人需要它......可能你想做的是改用onErrorContinue。请注意，此方法将从通量中删除错误项并继续处理，因此请务必牢记这一点。如果你想要错误+项目并停止一切，我相信如果你从这种方法中重新抛出异常，并且没有其他下游onErrorContinue，它应该终止通量。这是我使用 kafka 反应式 API 的一个例子

此外，传递给 onErrorContinue 的项将不是通量中的原始项，而是在反应链中的步骤中尝试处理的对象。例如，如果在尝试转换 B -> C 时链失败，但在转换 A -> B 后，您将在处理程序中获得的对象将是 B。

@EventListener(ApplicationReadyEvent.class)
public void process() {
receiver.receiveAutoAck()
.concatMap(Function.identity())
.flatMap(this::deserialzeRecord)
.flatMap(this::processEvent)
.flatMap(responseSender::send)
.onErrorContinue(this::handleRecordError)
.subscribe();
}
private void handleRecordError(Throwable throwable, Object record) {
log.error("Received error processing record={}", record, throwable);
}

如果您想终止助焊剂并删除其余项目，只需执行

private void handleRecordError(Throwable throwable, Object record) {
log.error("Received error processing record={}", record, throwable);
throw throwable;
}

Answer 2

如果某个值导致异常，则应将该值视为异常原因。

由你决定是否值得添加新的异常类型，

@RequiredArgsConstructor
public class WrongValueInStreamException extends RuntimeException {
@Getter
private final Object wrongValue;
}
public class StreamProcessor {
public String process(Integer value) {
if (value == 4) {
throw new WrongValueInStreamException(4);
}
return "processed " + value;
}
}

但只要异常传达有用且相关的信息，这是一个很好的做法：

.doOnError(WrongValueInStreamException.class::isInstance, e -> {
final Object value = ((WrongValueInStreamException) e).getWrongValue();
// use 'value'
})