为链中的每个处理程序分配不同的错误处理程序

Question

我有一个应用程序指定一个简单的流程，其中需要以特定顺序调用一系列服务来执行某些功能。 Spring 集成及其 DSL 似乎适合这种情况。

任何这些单独的服务都可以失败（这些服务是对外部应用程序的调用），此时链应该停止。不仅如此，控制流还需要根据失败的服务重定向到其他地方。换句话说，我需要为每个服务使用不同的错误处理程序。

问题是我很难找到如何做到这一点，如果有人能指出我正确的方向，我将不胜感激。到目前为止，这就是我解决问题的方式：

@Configuration
@EnableAutoConfiguration
@IntegrationComponentScan
public class Application {

public static void main(String[] args) throws InterruptedException {
    ConfigurableApplicationContext ctx =
            SpringApplication.run(Application.class, args);

    System.out.println(ctx.getBean(MySource.class).send("foo"));

    ctx.close();
}

@MessagingGateway
public interface MySource {

    @Gateway(requestChannel = "flow.input")
    String send(String string);

}

@Bean
public IntegrationFlow flow() {
    return f -> f
            .handle(String.class, (i, map) -> {
 //             if (true){
 //                 throw new RuntimeException();
 //             }
                System.out.println(i);
                return i + "a";
            })
            .handle(String.class, (i, map) -> {
                System.out.println(i);
                return i + "b";
            })
            .handle(String.class, (i, map) -> {
                System.out.println(i);
                return i + "c";
            });
   }
}

我一直在寻找如何为每个处理程序设置错误处理程序/通道，但没有发现任何有用的东西。

我几乎放弃了搜索并开始编写自己的代码（sigh），它看起来像这样：

public class FlowHarness {

    public static void main(String[] args) {
        Flow flow = FlowBuilder.flow()
            .nextStep(i -> ((Integer)i) + 1, e -> System.out.println("failed at step 1"))
            .nextStep(i -> ((Integer)i) + 1, e -> System.out.println("failed at step 2"))
            .nextStep(i -> ((Integer)i) + 1, e -> System.out.println("failed at step 3"))
            .build();

        flow.execute(new Integer(1));
    }
}

.nextStep() 的左侧参数是处理程序，而右侧参数是错误处理程序。它非常简单，但我希望它能说明这一点。

尽管所有现实生活中的流程都可以写在一个大类中，但我认为将每个子流程分解为单独的私有方法/类会更方便。非常感谢。

Answer 1

.handle()（以及任何其他消费者端点）可以通过 ExpressionEvaluatingRequestHandlerAdvice 提供：http://docs.spring.io/spring-integration/docs/4.3.9.RELEASE/reference/html/messaging-endpoints-chapter.html#expression-advice

@Bean
public IntegrationFlow advised() {
    return f -> f.handle((GenericHandler<String>) (payload, headers) -> {
        if (payload.equals("good")) {
            return null;
        }
        else {
            throw new RuntimeException("some failure");
        }
    }, c -> c.advice(expressionAdvice()));
}

@Bean
public Advice expressionAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setSuccessChannelName("success.input");
    advice.setOnSuccessExpressionString("payload + ' was successful'");
    advice.setFailureChannelName("failure.input");
    advice.setOnFailureExpressionString(
            "payload + ' was bad, with reason: ' + #exception.cause.message");
    advice.setTrapException(true);
    return advice;
}

@Bean
public IntegrationFlow success() {
    return f -> f.handle(System.out::println);
}

@Bean
public IntegrationFlow failure() {
    return f -> f.handle(System.out::println);
}

Answer 2

由于您希望链在失败时停止执行。我建议您可以使用带有适当处理程序的自定义异常