使用spring DSL构建spring集成发布策略

Question

我对Spring Integration是新手。我尝试使用文件拆分器将消息从文件中拆分出来，然后使用.aggregate（）构建单个消息并发送到输出通道。我有标记为true，因此apply-sequence现在默认为false。我已经使用EnricHeaders将correlationId设置为常量“1”。我在设置realease策略时遇到了困难，因为我没有在序列结束上保持。下面是我的代码的外观。

    IntegrationFlows
                .from(s -> s.file(new File(fileDir))
                                .filter(getFileFilter(fileName)),
                        e -> e.poller(poller))

                .split(Files.splitter(true, true)
                                .charset(StandardCharsets.US_ASCII),
                        e -> e.id(beanName)).enrichHeaders(h -> h.header("correlationId", "1"));

 IntegrationFlow integrationFlow = integrationFlowBuilder
            .<Object, Class<?>>route(Object::getClass, m -> m
                    .channelMapping(FileSplitter.FileMarker.class, "markers.input")
                    .channelMapping(String.class, "lines.input"))
            .get();

@Bean
    public IntegrationFlow itemExcludes() {
        return flow -> flow.transform(new ItemExcludeRowMapper(itemExcludeRowUnmarshaller)) //This maps each line to ItemExclude object
                .aggregate(aggregator -> aggregator
                        .outputProcessor(group -> group.getMessages()
                        .stream()
                        .map(message -> ((ItemExclude) message.getPayload()).getPartNumber())
                        .collect(Collectors.joining(","))))
                .transform(Transformers.toJson())
                .channel(customSource.itemExclude());
    }

    @Bean
    public IntegrationFlow itemExcludeMarkers() {
        return flow -> flow
                .log(LoggingHandler.Level.INFO)
                .<FileSplitter.FileMarker>filter(m -> m.getMark().equals(FileSplitter.FileMarker.Mark.END))
                .<FileHandler>handle(new FileHandler(configProps))
                .channel(NULL_CHANNEL);
    }

任何帮助都很感激。

Answer 1

我将在拆分器之前为correlationid移动头部enricher并使其如下所示：

 .enrichHeaders(h -> h
        .headerFunction(IntegrationMessageHeaderAccessor.CORRELATION_ID, 
                   m -> m.getHeaders().getId())) 

常量CorrelationID在多线程环境中绝对不好：不同的线程拆分不同的文件并将不同的行发送到同一个聚合器。因此，使用“1”作为关联键，您将始终拥有一个要聚合和释放的组。默认的序列行为是将原始消息ID填充到CorrelationID。由于您将不依赖filesplitter中的applysequence，所以我建议使用简单的解决方案来模拟这种行为。

正如Gary在回答中指出的，您需要考虑自定义releaseStrategy，并将filesplitter.fileMarker发送到聚合器。filesplitter.fileMarker.end具有lineCount属性，可以将该属性与messageGroup.size进行比较，以决定是否可以释放组。在生成输出结果的过程中，MessageGroupProcessor确实必须过滤FilespLitter.FileMarker消息。