春云流处理器 单输入多行输出

Question

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

输入:

{
    "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？

尽管如此，Spring Cloud Stream输出只是一个单独的事件，不适合我上面的案例，无法为Kafka生成3个事件。

谢谢你。

Answer 1

您生成的是一个有效的JSON数组。您试图生成的不是有效的JSON。但是只要你同意，你就可以简单地使用StringBuilder并把每个< i>book.toString()附加到它上面(而不是JSONArray)。那至少会产生你要找的东西。

此外，这确实不是一个与Spring Cloud Stream相关的问题，而是一个普通的Java/JSON问题，所以我认为StackOverflow上有更多特定的论坛可以为您提供更好的答案。

Answer 2

的确，我误解了。所以在本例中，我建议引入Spring集成和企业集成模式。你基本上对Splitter有一个明确的理由。有很多例子，但这里有一个简短的片段：@Splitter public List