java - Spring XD 流中的异常处理

Question

如何创建一个故障安全 Spring XD 流，该流将在一条特定消息触发异常后保持正常运行(即记录错误但继续使用流中的下一条消息)，而不必在每个流步骤中添加 try catch(Throwable)？

有没有简单的方法可以使用 Reactor 或 RxJava 模型来做到这一点？

使用 Reactor 的示例流:

@Override
public Publisher<Tuple> process(Stream<GenericMessage> inputStream) {
  return inputStream
      .flatMap(SomeClass::someFlatMap)
      .filter(SomeClass::someFilter)
      .when(Throwable.class, t -> log.error("error", t));
}

Answer 1

RxJava 可由处理器模块使用。创建时需要创建订阅并处理错误，订阅者需要添加 onError 处理程序:

       subject = new SerializedSubject(PublishSubject.create());
        Observable<?> outputStream = processor.process(subject);
        subscription = outputStream.subscribe(new Action1<Object>() {
            @Override
            public void call(Object outputObject) {
                if (ClassUtils.isAssignable(Message.class, outputObject.getClass())) {
                    getOutputChannel().send((Message) outputObject);
                } else {
                    getOutputChannel().send(MessageBuilder.withPayload(outputObject).build());
                }
            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                logger.error(throwable.getMessage(), throwable);
            }
        }, new Action0() {
            @Override
            public void call() {
                logger.error("Subscription close for [" + subscription + "]");
            }
        });

查看更多示例:https://github.com/spring-projects/spring-xd/tree/master/spring-xd-rxjava/src