春季集成文件分解器内存使用

Question

1. 文件对象上的split()操作是否缓冲并执行每一行，或者所有行一次加载到内存中？这是为了了解文件恰好包含 100,000 多行时的内存使用情况。

2. transformer 返回 void 公平吗？用途是从有效负载和负载中计算一些逻辑。 headers 然后将计算出的值添加到 headers 中。有更好的办法吗？

谢谢

更新：

return IntegrationFlows.from(fileReadingMessageSource(), p -> p.poller(pollerSpec()))
                        .enrichHeaders(Collections.singletonMap(ERROR_CHANNEL, appErrorChannel))
                        .split() // process file by file
                        .log(INFO, message -> "Started File: " + message.getHeaders().get("file_name"))
                        .enrichHeaders(h -> h.headerFunction("foo", m -> integrationUtil.constructFoo())) // fooobject
                        .split(fileSplitterSpec()) // split file lines
                            .filter(payload -> !(payload instanceof FileSplitter.FileMarker), e -> e.discardChannel("aggregatorChannel"))
                            .log(INFO, message -> "Payload: " + message.getPayload())
                            .transform(barTransformer)
                        .channel("aggregatorChannel")
                        .aggregate(new FileAggregator())
                        .log(INFO, message -> "Completed File: " + message.getHeaders().get("file_name"))
                        .aggregate()
                        .log(INFO, message -> "All Files Processed")
                        // .handle(null)
                        .get();

Answer 1

是的，这确实是 FileSplitter 的目的。它的内部逻辑基于 FileIterator，它逐行读取并将其发送到分离器输出通道。

不，变压器不能返回void。这不是它的目的。听起来更像 enrichHeaders() 更适合您。消息是不可变的，您无法修改当前消息以获得可能的进一步逻辑。您使用新数据（或标头）构建一条新消息，并将其作为流下游的回复发出。