java - Spring Cloud流处理器单输入，多行输出

Question

我一直在尝试从 kafka 流式传输我的 json 事件，将其压平，然后使用 Spring Cloud Stream 将其推回到另一个主题。

输入:

{
    "major": "Computer Science",
    "books": [{
        "title": "Learn C",
        "author": "Prof C"
    },
    {
        "title": "Learn Java",
        "author": "Java Expert"
    },{
        "title": "Learn Python",
        "author": "Python Master"
    },]
}

压平过程:

@ServiceActivator(inputChannel = Sink.INPUT, outputChannel = Source.OUTPUT)
public String(String event){
    JSONArray result = new JSONArray();

    JSONObject rawEvent = new JSONObject(event);

    String major = rawEvent.get("major");
    JSONArray books = rawEvent.get("books");

    for (int i = 0; i < books.length; i++){
        JSONObject book = books.get(i);
        book.put("major", major);
        result.put(book)
    }

    return result.toString();
}

它只产生:

    [{"major":"Computer Science", "books.title":"Learn C", "books.author":"Prof C"}, 
{"major":"Computer Science", "books.title":"Learn Java", "books.author":"Java Expert"}, 
{"major":"Computer Science", "books.title":"Learn Python", "books.author":"Python Master"}]

我的问题是如何让它变成这样

{"major":"Computer Science", "books.title":"Learn C", "books.author":"Prof C"}
{"major":"Computer Science", "books.title":"Learn Java", "books.author":"Java Expert"}
{"major":"Computer Science", "books.title":"Learn Python", "books.author":"Python Master"}

所以我可以在多个 JSONObject 中推回，而不是像我所做的那样推回单个 JSONArray？

Afaik，Spring Cloud Stream 输出只是一个事件，不适合我上面的情况，无法向 Kafka 生成 3 个事件。

谢谢。

Answer 1

确实是我理解错了。所以在这种情况下我建议引入 Spring Integration和 Enterprise Integration Patterns 。对于 Splitter 来说，你基本上已经有了一个明确的案例。那里有很多例子，但这里是一个快速片段:
@Splitter
public List<String> split(String input) {
// basically split your input into a collection and splitter will send out each element as a separate message
}
希望有帮助