SCDF处理器输出对象数组，RabbitMQ绑定器下如何适配单条目Sink？

首先明确：这个方案完全可行，Spring Cloud Stream的RabbitMQ绑定器原生支持集合消息拆分，下面给两种落地方式：

方式1：用绑定器自动拆分（最省心） #

你现在的处理器代码不用改，只需要加一行配置就行：
在你的处理器应用配置文件（application.yml/application.properties）里添加：

spring.cloud.stream.bindings.output.producer.collection-mode=split

这个配置告诉RabbitMQ绑定器：当输出是集合类型时，自动把每个元素拆成单独的消息发出去。这样你的Sink就能逐个收到XYZObject（注意：你当前Sink接收的是XYZInput，这里存在类型不匹配问题，要么把Sink改成接收XYZObject，要么在处理器里把XYZObject转成XYZInput，否则会触发消息转换异常）。

原处理器代码保持不变：

@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public List<XYZObject> getAll(XYZInput inp) {
    List<XYZObject> xyzs = dbService.findAllByDataType(inp.getDataType());
    return xyzs;
}

方式2：手动拆分（更灵活） #

如果需要添加自定义消息头、做复杂类型转换，直接在处理器里手动遍历列表发送单条消息即可：

@Autowired
private MessageChannel output; // 对应Processor.OUTPUT通道

@StreamListener(Processor.INPUT)
public void getAll(XYZInput inp) {
    List<XYZObject> xyzs = dbService.findAllByDataType(inp.getDataType());
    xyzs.forEach(xyz -> {
        // 若Sink需要XYZInput类型，在此处添加转换逻辑
        // XYZInput convertedInput = convertXYZObjectToXYZInput(xyz);
        output.send(MessageBuilder.withPayload(xyz).build());
    });
}

这种方式完全由代码控制拆分逻辑，适配各种特殊业务场景。

关键提醒 #

你当前的Sink接收XYZInput，但处理器返回的是XYZObject列表，这是硬类型不兼容问题，必须先解决：要么修改Sink的@StreamListener参数为XYZObject，要么在处理器里把每个XYZObject转换成XYZInput再发送，否则Sink会因为消息类型不匹配抛出异常。

内容的提问来源于stack exchange，提问作者Suranjan Poudel