Splitter在异常期间中止处理后续消息

Question

我有一个要求，就是把消息分开，然后一个接一个地处理。如果有任何消息失败，我会将其报告给错误通道，然后继续处理下一个可用的消息。

我使用的是带有1.0.0-快照的流启动程序

我用拆分器编写了一个示例程序

     @Bean
      public MessageChannel channelSplitOne() {
        return new DirectChannel();
      }


      @StreamListener(INTERNAL_CHANNEL)
      public void channelOne(String message) {
        if (message.equals("l")) {
          throw new RuntimeException("Error due to l");
        }
        System.out.println("Internal: " + message);
      }


      @Splitter(inputChannel = Sink.INPUT, outputChannel = INTERNAL_CHANNEL)
      public List<Message> extractItems(Message<String> input) {
        return Arrays.stream(input.getPayload().split(""))
            .map(s -> MessageBuilder.withPayload(s).copyHeaders(input.getHeaders()).build())
            .collect(Collectors.toList());
      }

当我以Hello的形式发送消息时，提示是“h”、“e”、“o”将被处理，而“l”将被报告为错误。

但在这里，在后面的'l'，处理没有恢复。

有什么办法可以做到这一点吗。

Answer 1

您可以这样做，但使用的是@ServiceActivator而不是@StreamListener。第一个选项具有adviceChain选项，您可以在其中注入一个ExpressionEvaluatingRequestHandlerAdvice：https://docs.spring.io/spring-integration/docs/5.0.4.RELEASE/reference/html/messaging-endpoints-chapter.html#expression-advice。

拆分器类似于Java中的常规循环的问题，因此要在错误之后继续，我们需要在那里添加一个try...catch。但这已经不是分裂的责任了。因此，我们必须把这样的逻辑移到我们有错误问题的地方。