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"}

所以我可以像我所做的那样将 mutilple JSONObject 推回而不是单个 JSONArray？

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

谢谢。

Answer 1

确实我误会了。 所以在这种情况下，我建议引入Spring Integration 和Enterprise Integration Patterns。您基本上对拆分器有一个明确的案例。那里有很多例子，但这里有一个快速的 sn-p： @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 } 希望有帮助

Answer 2

您生成的是一个有效的 JSON 数组。您尝试生成的不是有效的 JSON。但是，只要您对此感到满意，您就可以简单地使用 StringBuilder 并将每个 book.toString() 附加到它（而不是 JSONArray）。这至少会产生你正在寻找的东西。

另外，这真的不是一个与 Spring Cloud Stream 相关的问题，而是一般的 Java/JSON 问题，所以我想 StackOverflow 上有更具体的论坛可以为您提供更好的答案。