具有循环以及多个HTTP和异步SQS的Spring Integration块流

问题描述

我有一个流程

1. Starts with a config map -> MainGateway.start(configMap) -> void
2. Splits map into multiple messages per entry
3. For every config entry do the following using an orchestrator java class:
   BEGIN LOOP (offset and limit)
      Data d = HTTPGateway.getData();
      PublishGateway.sendMessage(d); -> Send to 2 SQS queues    
   END LOOP

要求 我必须通过cron安排此流程。一种选择是提供将启动该流程的HTTP端点。但是，第二个HTTP请求应该等待/超时/错误，直到第一个HTTP请求完成。

问题 我一直在寻找对流线程实施阻塞的障碍，直到它完成并且只有一个线程http处理器，所以一次只处理1个请求，我就知道流何时完成。 （对于所有配置条目对象，LOOP结束，并且到SQS的所有消息都被确认）。我该如何实现？我有一个循环，并且正在使用带有执行程序的pub-sub通道进行并行配置和并行SQS调度。

为清楚起见，我已经修剪了下面的XML config。

   <!-- Bring in list of Configs to process -->
    <int:gateway service-interface="Gateway"
                 default-request-channel="configListChannel" />

    <int:chain input-channel="configListChannel" output-channel="configChannel">
        <!-- Split the list to one instance of config per message -->
        <int:splitter/>
        <int:filter expression="payload.enablePolling" />
    </int:chain>

    <!-- Manually orchestrate a loop to query a system as per config and publish messages to SQS -->
    <bean class="orchestrator" id="orchestrator" />
    <int:service-activator ref="orchestrator" method="getData" input-channel="configChannel" />

    <!-- The flow from this point onwards is triggered inside a loop controlled by the orchestrator
         The following Gateway calls are inside orchestrators loop -->

    <!-- Create a Http request from the orchestrator using a Gateway -->
    <int:gateway service-interface="HttpGateway">
        <int:method name="getData"
                    request-channel="requestChannel"
                    payload-expression="#args[0]">
        </int:method>
    </int:gateway>

    <!-- Transform request object to json and invoke Http endpoint -->
    <int:chain input-channel="requestChannel" id="httpRequestChain">
        <int:object-to-json-transformer />
        <int-http:outbound-gateway url-expression="headers['config'].url"
                                   http-method="POST"
                                   expected-response-type="java.lang.String"
        />
    </int:chain>

    <!-- Publish Messages to Outbound Gateway -->
    <task:executor id="executor" pool-size="5" />
    <int:publish-subscribe-channel id="publishChannel" task-executor="executor" />
    <int:gateway service-interface="PublishGateway" >
        <int:method name="publishToOutbound" payload-expression="#args[0]" request-channel="publishChannel" />
    </int:gateway>


    <!-- Route to System A SQS with transformations (omitted here)-->
    <int-aws:sqs-outbound-channel-adapter sqs="amazonSQS" channel="publishChannel" queue="system-a-queue" success-channel="successChannel" failure-channel="errorChannel"/>

    <!-- Route to System B SQS with transformations (omitted here)-->
    <int-aws:sqs-outbound-channel-adapter sqs="amazonSQS" channel="publishChannel" queue="system-b-queue" success-channel="successChannel" failure-channel="errorChannel"/>

    <int:logging-channel-adapter logger-name="sqsCallbackLogger" log-full-message="true" channel="successChannel" />

同时，我正在尝试将spring-integration-samples的A B C障碍示例适应我的用例。

解决方法

正如您在评论中指出的那样，可以在解决方案中使用aggregator方法。

这样，您可以汇总这些并行SQS请求的结果，并在原始请求者中等待汇总答复。这样，即使您的流程内部仍然并发，也确实会阻止它。您呼叫一个网关，对此的答复将来自聚合器。