Spring Cloud Stream 在发送后处理消息

Question

在 Spring Cloud Stream 的新版本中，@EnableBinding 和声明式编程风格将被弃用。使用函数式编程风格，如何处理使用管道消息处理的？

第一种情况

我需要处理发送到输出通道的消息，例如记录成功处理，或将它们存储到数据库。

@EnableBinding(Processor.class)
public class MessageProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageProcessor.class);
    private final MessageChannel output;

    public MessageProcessor(MessageChannel output) {
        this.output = output;
    }

    @StreamListener(Processor.INPUT)
    public void process(Message<String> message) {
        LOGGER.info("Receive message: {}", message);
        output.send(message);
        /*Do some work with message here*/
        LOGGER.info("Finish processing for message: {}", message);
    }
}

第二种情况

我收到一条消息，存储 DTO 的集合，我需要分别处理每个 DTO 对象。

@EnableBinding(Processor.class)
public class BatchMessageProcessor {
    private final MessageChannel output;

    public BatchMessageProcessor(MessageChannel output) {
        this.output = output;
    }

    @StreamListener(Processor.INPUT)
    public void process(Message<PackageDto> pgk) {
        Stream.ofNullable(pgk)
                .map(Message::getPayload)
                .map(PackageDto::getMessages)
                .flatMap(Collection::stream)
                .filter(Objects::nonNull)
                /*Sent messages separately*/
                .forEach(m -> output.send(MessageBuilder.withPayload(m).build()));
    }
}

如何在 Spring Cloud Stream 中使用函数式编程风格来处理这种情况？

Answer 1

使用StreamBridge 或反应器EmitterProcessor。

见Sending arbitrary data to an output。