我正在使用 Spring Boot 1.5.13.RELEASE 和 Spring Integration 4.3.16.RELEASE 开发一个应用程序。
我对 Spring Integration 还很陌生，并且遇到了一个问题。
所以基本的想法是，在一些外部触发器（可能是 HTTP 调用）上，我需要创建一个 IntegrationFlow ，它将使用来自rabbitMQ队列的消息，对它们进行一些处理，然后（可能）生成到另一个rabbitMQ端点。
现在这应该会发生很多次，所以我必须创建multiple集成流程。
我在用集成流程上下文像这样注册每个 IntegrationFlow：
IntegrationFlowContext flowContext;
...
IntegrationFlow integrationFlow = myFlowFactory.makeFlow(uuid);
...
flowContext.registration(integrationFlow).id(callUUID).register();
我必须澄清这可以同时发生，创建multiple集成同时进行。
因此，每次我尝试创建 IntegrationFlow 时，我的“源”都是一个网关，如下所示：
MessagingGatewaySupport sourceGateway = Amqp
.inboundGateway(rabbitTemplate.getConnectionFactory(), rabbitTemplate, dynamicQueuePrefix+uuid)
.concurrentConsumers(1)
.adviceChain(retryInterceptor)
.autoStartup(false)
.id("sgX-" + uuid)
.get();
它还不是 @Bean，但我希望它在注册每个 IntegrationFlow 时注册。
我的“目标”是 Amqp OutBoundAdapter，如下所示：
@Bean
public AmqpOutboundEndpoint outboundAdapter(
RabbitTemplate rabbitTemplate,
ApplicationMessagingProperties applicationMessagingProperties
) {
return Amqp.outboundAdapter(rabbitTemplate)
.exchangeName("someStandardExchange")
.routingKeyExpression("headers.get('rabbitmq.ROUTING_KEY')")
.get();
}
现在这个IS已经有一个 bean，每次我尝试创建流时都会注入它。
我的流程如下所示：
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();
}
我的问题是，虽然应用程序启动正常并创建第一个 IntegrationFlows 等。后来，我遇到了这种异常：
java.lang.IllegalStateException：无法在bean名称“org.springframework.integration.transformer.MessageTransformingHandler#872”下注册对象[org.springframework.integration.transformer.MessageTransformingHandler#872]：已经存在对象[org.springframework.integration .transformer.MessageTransformingHandler#872] 绑定
我什至尝试为每个使用的组件设置一个 id，它应该用作 beanName ，如下所示：
.transform(Transformers.fromJson(HashMap.class, jsonObjectMapper), tf -> tf.id("tf1-"+uuid))
但是，即使像 .filter 这样的组件的 bean 名称问题已经解决，我仍然得到关于 MessageTransformingHandler 的相同异常。
UPDATE
我没有提到这样一个事实：每一次IntegrationFlow工作完成后，使用以下命令将其删除IntegrationFlowContext像这样：
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.
这并没有我想要的那么快发生。
但我想这是预期的，因为每次线程获取锁时，都需要一些时间register流程及其所有组成部分或注销流程及其所有组成部分。
