spring集成dsl kafkaproducercontext配置

daolsyd0  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(389)

我正在尝试改编以下示例:https://github.com/joshlong/spring-and-kafka
使用以下库的最新稳定版本:

org.apache.kafka > kafka_2.10 > 0.8.2.2
org.springframework.integration > spring-integration-kafka > 1.2.1.RELEASE
org.springframework.integration > spring-integration-java-dsl > 1.1.0.RELEASE

集成dsl库似乎经历了一次重构,这可能是由新kafkaproducer的引入所驱动的。
以下是我的生产者配置代码:

@Bean(name = OUTBOUND_ID)
IntegrationFlow producer() {
    log.info("starting producer flow..");

    return flowDefinition -> {
        ProducerMetadata<String, String> getProducerMetadata = new ProducerMetadata<>(this.kafkaConfig.getTopic(),
                    String.class, String.class, new StringSerializer(), new StringSerializer());

        KafkaProducerMessageHandler kafkaProducerMessageHandler = Kafka.outboundChannelAdapter(props ->
                props.put("timeout.ms", "35000"))
                .messageKey(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
                .addProducer(getProducerMetadata, this.kafkaConfig.getBrokerAddress())
                .get();

        flowDefinition
                .handle(kafkaProducerMessageHandler);
    };
}

以及生成消息的代码:

@Bean
@DependsOn(OUTBOUND_ID)
CommandLineRunner kickOff(@Qualifier(OUTBOUND_ID + ".input") MessageChannel in) {
    return args -> {              
        for (int i = 0; i < 1000; i++) {
            in.send(MessageBuilder.withPayload("#" + i).setHeader(KafkaHeaders.TOPIC, this.kafkaConfig.getTopic()).build());
                log.info("sending message #" + i);
        }
    };
}

这是我得到的例外:

Caused by: org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler#0]; nested exception is java.lang.NullPointerException
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:84)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:287)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:245)
at jc.DemoApplication$ProducerConfiguration.lambda$kickOff$0(DemoApplication.java:104)
at org.springframework.boot.SpringApplication.runCommandLineRunners(SpringApplication.java:673)
... 10 more
Caused by: java.lang.NullPointerException
at org.springframework.integration.kafka.support.KafkaProducerContext.getTopicConfiguration(KafkaProducerContext.java:67)
at org.springframework.integration.kafka.support.KafkaProducerContext.send(KafkaProducerContext.java:201)
at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleMessageInternal(KafkaProducerMessageHandler.java:88)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
... 18 more

更新:
完整的工作源代码可以在我的fork中找到:
https://github.com/magiccrafter/spring-and-kafka

mdfafbf1

mdfafbf11#

抱歉耽搁了。
你的问题就在早上 IntegrationComponentSpec 示例化:

KafkaProducerMessageHandler kafkaProducerMessageHandler = Kafka.outboundChannelAdapter(props ->
          props.put("timeout.ms", "35000"))
          .messageKey(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
          .addProducer(getProducerMetadata, this.kafkaConfig.getBrokerAddress())
          .get();

你不应该打电话 .get() 你自己。这个 KafkaProducerMessageHandlerSpecComponentsRegistration 只有si-javadsl才能正确解析。代码如下所示:

public Collection<Object> getComponentsToRegister() {
    this.kafkaProducerContext.setProducerConfigurations(this.producerConfigurations);
    return Collections.<Object>singleton(this.kafkaProducerContext);
}

因为这个代码没有被调用 this.producerConfigurations 未填充到 this.kafkaProducerContext . 尽管最后一个必须注册为bean。
所以,要解决你的问题,你应该只处理 IntegrationComponentSpec 在dsl定义中。
只需获得 KafkaProducerMessageHandlerSpec 把它用来 .handle() 在下面。如果我们可以使用 Kafka.outboundChannelAdapter() 直接从 .handle() .

相关问题