使用具有 spring 集成的多个 KafkaProducerMessageHandler 时出现问题

Question

这是 post MaprStream with spring integration Kafka Producer issue

的扩展

在向 maprstream

发布消息时尝试实现同步 属性 时，我遇到了多个 KafkaProducerMessageHandlers 的问题

@Autowired
Qualifier("abcHandler.handler")
private KafkaProducerMessageHandler abcHandler;

@Autowired
Qualifier("xyzHandler.handler")
private KafkaProducerMessageHandler xyzHandler;

@PostConstruct
public void init() {
    this.abcHandler.setSync(true);
    this.xyzHandler.setSync(true);
}

Bean 配置：

<int:chain input-channel="inputToKafka"> 
    <int-kafka:outbound-channel-adapter 
            id="abcHandler" 
            kafka-template="template" 
            topic="${maprstream.stream.topicname}" > 
    </int-kafka:outbound-channel-adapter> 
</int:chain>

<int:chain input-channel="inputToKafka1"> 
    <int-kafka:outbound-channel-adapter 
            id="xyzHandler" 
            kafka-template="template1" 
            topic="${maprstream.stream.topicname1}" > 
    </int-kafka:outbound-channel-adapter> 
</int:chain>

我在尝试加载 bean 时遇到以下异常。

Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'maprStreamProducerHandlerSync': Unsatisfied dependency expressed through field 'abcHandler'; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {@org.springframework.beans.factory.annotation.Autowired(required=true), @org.springframework.beans.factory.annotation.Qualifier(value=abcHandler.handler)}

有人可以帮我解决这个问题吗？

Answer 1

为什么 <chain/> 中有这些单一组件？链通常用于对多个元素进行分组。

参见the documentation about chains；向下滚动到 'id' 属性`.

<int:chain id="chain1" input-channel="inputToKafka"> 
    <int-kafka:outbound-channel-adapter 
            id="abcHandler" 
            kafka-template="template" 
            topic="${maprstream.stream.topicname}" > 
    </int-kafka:outbound-channel-adapter> 
</int:chain>

链中的组件获得复合 bean 名称。在这种情况下，它将是 chain1$child.abcHandler.handler。自动接线时，当您有多个适配器时，您需要在 @Qualifier 中使用此值。