Mule中的并行处理:获得正确响应的问题

Question

要求是开发mule流并行调用3个不同的同步服务,然后汇总每个服务器的响应并将其发送回调用者.

我遵循了文档中提到的fork join方法以及如何进行并行出站调用.我的配置文件如下所示:

            <flow name="fork">
            <http:inbound-endpoint host="localhost" port="8090" path="mainPath" exchange-pattern="request-response">
            <set-property propertyName="MULE_CORRELATION_GROUP_SIZE"
                value="2" />
            <all enableCorrelation="IF_NOT_SET">
                <async>
                    <set-property propertyName="MULE_CORRELATION_SEQUENCE"
                        value="1" />
                    <flow-ref name="parallel1" />
                </async>
                <async>
                    <set-property propertyName="MULE_CORRELATION_SEQUENCE"
                        value="2" />
                    <flow-ref name="parallel2" />
                </async>
            </all>
        </flow>

        <sub-flow name="parallel1">
            <logger level="INFO" message="parallel1: processing started" />
            <!- Transformation payload -->
            <http:outbound-endpoint address="..."
                exchange-pattern="request-response" />
            <logger level="INFO" message="parallel1: processing finished" />
            <flow-ref name="join" />
        </sub-flow>

        <sub-flow name="parallel2">
            <logger level="INFO" message="parallel2: processing started" />
            <!- Transformation payload -->
            <http:outbound-endpoint address="..."
                exchange-pattern="request-response" />
            <logger level="INFO" message="parallel2: processing finished" />
            <flow-ref name="join" />
        </sub-flow>

        <sub-flow name="join">
            <collection-aggregator timeout="6000"
                failOnTimeout="true" />
            <combine-collections-transformer />
            <logger level="INFO"  message="Continuing processing of: #[message.payloadAs(java.lang.String)]" />
            <set-payload value="Soap XML Response"/>
        </sub-flow>

我能够验证,直到"join"子流工作正常,但响应不会回复为"Soap XML Response".响应是相同的初始SOAP请求.

如何让这个线程等到子流处理完成后,无论"join"子流返回什么,它都会发回响应？

Answer 1

上面帖子中的fork连接看起来不错.这里的问题是在连接之后无法捕获有效负载并将其带回主流.

当并行调用异步时,主流继续而不等待连接输出.

我已经修改了流程来解决这个问题.现在流程将有一个处理器等待回复并读取连接的输出以写入http转换器.

    <flow name="fork">
        <http:inbound-endpoint host="localhost" port="8090" path="mainPath" exchange-pattern="request-response">
            <!-- To get back the response after the fork-join -->
        <request-reply timeout="60000">
            <jms:outbound-endpoint queue="parallel.processor.queue">
                <message-properties-transformer scope="outbound">
                    <delete-message-property key="MULE_REPLYTO" />
                </message-properties-transformer>
            </jms:outbound-endpoint>
            <jms:inbound-endpoint queue="join.queue" >      
            </jms:inbound-endpoint>
        </request-reply>            
    </flow>

    <flow name="fork_join_flow" >
        <jms:inbound-endpoint queue="parallel.processor.queue" exchange-pattern="one-way" />        
        <set-property propertyName="MULE_CORRELATION_GROUP_SIZE"
                value="2" />
        <all enableCorrelation="IF_NOT_SET">
            <async>
                <set-property propertyName="MULE_CORRELATION_SEQUENCE"
                    value="1" />
                <flow-ref name="parallel1" />
            </async>
            <async>
                <set-property propertyName="MULE_CORRELATION_SEQUENCE"
                    value="2" />
                <flow-ref name="parallel2" />
            </async>
        </all>
    </flow>

    <sub-flow name="parallel1">
        <logger level="INFO" message="parallel1: processing started" />
        <!- Transformation payload -->
        <http:outbound-endpoint address="..."
            exchange-pattern="request-response" />
        <logger level="INFO" message="parallel1: processing finished" />
        <flow-ref name="join" />
    </sub-flow>

    <sub-flow name="parallel2">
        <logger level="INFO" message="parallel2: processing started" />
        <!- Transformation payload -->
        <http:outbound-endpoint address="..."
            exchange-pattern="request-response" />
        <logger level="INFO" message="parallel2: processing finished" />
        <flow-ref name="join" />
    </sub-flow>

    <sub-flow name="join">
        <collection-aggregator timeout="6000"
            failOnTimeout="true" />
        <combine-collections-transformer />
        <logger level="INFO"  message="Continuing processing of: #[message.payloadAs(java.lang.String)]" />
        <set-payload value="Soap XML Response"/>
        <jms:outbound-endpoint queue="join.queue">              
        </jms:outbound-endpoint>
    </sub-flow>

希望这可以帮助.