我正在使用 Spring Boot 1.5.13.RELEASE 和 Spring Integration 4.3.16.RELEASE 开发应用程序。
我对 Spring Integration 很陌生，遇到了一个问题。
所以基本的想法是，在一些外部触发器（可能是和 HTTP 调用）上，我需要创建一个 IntegrationFlow，它将使用来自 rabbitMQ 队列的消息，使用它们做一些工作，然后（可能）生成到另一个 rabbitMQ 端点。
现在这应该发生很多次，所以我将不得不创建多个集成流。
我正在使用IntegrationFlowContext来注册每个 IntegrationFlow，如下所示：
IntegrationFlowContext flowContext;
...
IntegrationFlow integrationFlow = myFlowFactory.makeFlow(uuid);
...
flowContext.registration(integrationFlow).id(callUUID).register();
我必须澄清这可以同时发生，同时创建多个集成流。
所以每次我尝试创建一个集成流时，我的“源”是一个看起来像这样的网关：
MessagingGatewaySupport sourceGateway = Amqp
.inboundGateway(rabbitTemplate.getConnectionFactory(), rabbitTemplate, dynamicQueuePrefix+uuid)
.concurrentConsumers(1)
.adviceChain(retryInterceptor)
.autoStartup(false)
.id("sgX-" + uuid)
.get();
它不是@Bean（还），但我希望它在每个 IntegrationFlow 注册时都能注册。
我的“目标”是一个 AmqpOutBoundAdapter，如下所示：
@Bean
public AmqpOutboundEndpoint outboundAdapter(
RabbitTemplate rabbitTemplate,
ApplicationMessagingProperties applicationMessagingProperties
) {
return Amqp.outboundAdapter(rabbitTemplate)
.exchangeName("someStandardExchange")
.routingKeyExpression("headers.get('rabbitmq.ROUTING_KEY')")
.get();
}
现在这个已经是一个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 的相同异常。
更新
我没有提到这样一个事实，即一旦每个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.
这并没有我想的那么快。
但我猜这是意料之中的，因为每次线程获取锁时，注册流及其所有组件或注销流及其所有组件都需要一些时间。