Spring integration Java DSL-动态创建IntegrationFlow

Question

我正在使用Spring Boot 1.5.13.release和Spring Integration 4.3.16.release开发一个应用程序。

我对Spring Integration相当陌生，我遇到了一个问题。

因此，基本思想是，在一些外部触发器（可能是和HTTP调用）上，我需要创建一个IntegrationFlow来使用来自rabbitMQ队列的消息，对它们进行一些处理，然后（可能）生成到另一个rabbitMQendpoint。

IntegrationFlowContext flowContext;
...
IntegrationFlow integrationFlow = myFlowFactory.makeFlow(uuid);
...
flowContext.registration(integrationFlow).id(callUUID).register();

我必须澄清，这可以同时发生，同时创建多个IntegrationFlow。

因此，每次我试图创建IntegrationFlow时，我的“源”是一个如下所示的网关：

MessagingGatewaySupport sourceGateway = Amqp
        .inboundGateway(rabbitTemplate.getConnectionFactory(), rabbitTemplate, dynamicQueuePrefix+uuid)
        .concurrentConsumers(1)
        .adviceChain(retryInterceptor)
        .autoStartup(false)
        .id("sgX-" + uuid)
        .get();

它还不是@bean，但我希望在注册每个IntegrationFlow时注册它。

@Bean
public AmqpOutboundEndpoint outboundAdapter(
        RabbitTemplate rabbitTemplate,
        ApplicationMessagingProperties applicationMessagingProperties
) {
    return Amqp.outboundAdapter(rabbitTemplate)
            .exchangeName("someStandardExchange")
            .routingKeyExpression("headers.get('rabbitmq.ROUTING_KEY')")
            .get();
}

public IntegrationFlow configure() {
    return IntegrationFlows
            .from(sourceGateway)
            .transform(Transformers.fromJson(HashMap.class, jsonObjectMapper))
            .filter(injectedGenericSelectorFilter)
            .<HashMap<String, String>>handle((payload, headers) -> {

                String uuid = payload.get("uuid");

                boolean shouldForwardMessage = myInjectedApplicationService.isForForwarding(payload);
                myInjectedApplicationService.handlePayload(payload);

                return MessageBuilder
                        .withPayload(payload)
                        .setHeader("shouldForward", shouldForwardMessage)
                        .setHeader("rabbitmq.ROUTING_KEY", uuid)
                        .build();
            })
            .filter("headers.get('shouldForward').equals(true)")
            .transform(Transformers.toJson(jsonObjectMapper))
            .handle(outboundAdapter)
            .get();
}

.transform(Transformers.fromJson(HashMap.class, jsonObjectMapper), tf -> tf.id("tf1-"+uuid))

但是，即使像.filter这样的组件的bean名称问题已经解决，我仍然会得到关于MessageTransFormingHandler的相同异常。

我没有提到这样一个事实，即一旦每个IntegrationFlow完成其工作，就会使用IntegrationFlow Context删除它，如下所示：

flowContext.remove(flowId);

因此，通过使用相同的对象作为锁来同步流注册块和流移除块，似乎已经（某种程度上）奏效了。

...
private final Object lockA = new Object();
...

public void appendNewFlow(String callUUID){
    IntegrationFlow integrationFlow = myFlowFactory.makeFlow(callUUID);

    synchronized (lockA) {
        flowContext.registration(integrationFlow).id(callUUID).register();
    }
}

public void removeFlow(String flowId){

    synchronized (lockA) {
        flowContext.remove(flowId); 
    }

}
...

...Waiting for workers to finish.
...
...Successfully waited for workers to finish.

但我想这是意料之中的，因为每次线程获取锁时，注册流及其所有组件或取消注册流及其所有组件都需要一些时间。

Answer 1

你还有这个：

.transform(Transformers.toJson(jsonObjectMapper))

如果您也在其中添加.id()将如何工作？

另一方面，既然您说这是并发发生的，那么我想知道是否可以将代码中的某些部分合并，例如包装flowcontext.registration（integrationFlow）.id（callUUID）.register（）；。