Spring集成使异步网关等待聚合结果

Question

null

public void add(Message<?> messageToAdd) {
        if(messageToAdd.getPayload() instanceof FileMarker) {
            FileMarker marker = (FileMarker) messageToAdd.getPayload();
            switch (marker.getMark()) {
                case START:
                    replyChannel = messageToAdd.getHeaders().get(MessageHeaders.REPLY_CHANNEL);
                    break;
                case END:
                    expectedRowCount.set(marker.getLineCount());
                    break;
                default:
                    throw new IllegalStateException("Unexpected file mark");
            }
        } else {
            ProcessingResult processingResult = messageToAdd.getHeaders()
                    .get(ProcessingConstants.PROCESSING_RESULT_HEADER, ProcessingResult.class);
            assert processingResult != null;
            rowCount.incrementAndGet();
            switch (processingResult) {
                case FILTERED:
                    filteredCount.incrementAndGet();
                    break;
                case PROCESSED:
                    processedCount.incrementAndGet();
                    break;
                case PROCESSING_ERROR:
                    processingErrorCount.incrementAndGet();
                    break;
                case KAFKA_ERROR:
                    kafkaErrorCount.incrementAndGet();
                    break;
                default:
                    throw new IllegalStateException("Unrecognized processing result: " + processingResult);
            }
        }
    }

重新创建问题的测试可以在https://github.com/hawk1234/spring-integration-example commit 9f121f0729d8076872e6fbdcd7b1b91ca9ea8cb4中找到。当您运行测试时，应用程序日志可以在路径build/logs/spring-integration-example.log下获得。当前测试挂起，因为网关从未收到响应。此外，整个流程都在进行中，所以只有在超时后组才会释放。

Answer 1

我已经创建了自定义消息组

那就有点不对劲了。

在group,standard group中已经有了所有消息（包括FileMarker)时，必须执行所有必需的逻辑。有一个MessageGroupProcessor，当ReleaseStrategy返回true时调用它。因此，这里有组的所有消息，您可以执行所有所需的统计信息收集，以产生所需的答复。