java - 使用 Spring 集成 java 配置创建消息驱动的入站 channel 适配器以连接到 Kafka

Question

使用 Spring Integration java dsl 创建 kafka 消费者的 Java 代码

Answer 1

使用下面的代码创建一个消息驱动适配器来连接以消费来自kafka的消息

@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.topic}")
private String springIntegrationKafkaTopic;
@Bean
public IntegrationFlow kafkaReader() throws Exception {

    return IntegrationFlows
            .from(Kafka.messageDrivenChannelAdapter(listener(),ListenerMode.record))
            .channel("queureader")
            .get();
}

@Bean
public KafkaMessageListenerContainer listener() {
    return new KafkaMessageListenerContainer(consumerFactory(), new ContainerProperties(this.springIntegrationKafkaTopic));
}

@Bean
public ConsumerFactory consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafkaListener");
    return new DefaultKafkaConsumerFactory(props);
}


@ServiceActivator(inputChannel = "queureader")
public void Print(Message<?> msg)  {

    System.out.println(msg.getPayload().toString());
}

<小时/>

在application.properties中

• spring.kafka.bootstrap-servers = 我们需要提及服务器名称-spring.kafka.topic=主题名称

您可以提及 ConsumerConfig.GROUP_ID_CONFIG 的任何值。