springboot 连接 kafka集群（kafka版本 2.13-3.4.0）_百度小程序

您的位置：上海毫米网络优化公司 > 小程序 > 百度小程序 >

springboot 连接 kafka集群

一、环境搭建

1.1 springboot 环境
1.2 kafka 依赖
二、 kafka 配置类

2.1 发布者

2.1.1 配置

2.1.2 构建发布者类
2.1.3 发布消息
2.2 消费者

2.2.1 配置
2.2.2 构建消费者类

2.2.3 进行消息消费

一、环境搭建

1.1 springboot 环境

JDK 11+

Maven 3.8.x+

springboot 2.5.4 +

1.2 kafka 依赖

springboot的pom文件导入

       
            org.springframework.kafka
            spring-kafka
        
        
            org.springframework.kafka
            spring-kafka-test
            test
        
        
            org.apache.kafka
            kafka-clients
            3.4.0

二、 kafka 配置类

2.1 发布者

2.1.1 配置

发布者我们使用 KafkaTemplate 来进行消息发布，所以需要先对其进行一些必要的配置。

@Configuration
@EnableKafka
public class KafkaConfig {
     /***** 发布者 *****/
    //生产者工厂
    @Bean
    public ProducerFactory producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    //生产者配置
    @Bean
    public Map producerConfigs() {
        Map props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.83:9092,192.168.2.84:9092,192.168.2.86:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }
    //生产者模板
    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

2.1.2 构建发布者类

配置完发布者，下来就是发布消息，我们需要继承 ProducerListener 接口，该接口完整信息如下：

public interface ProducerListener {
    void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata);
    void onError(ProducerRecord producerRecord, RecordMetadata recordMetadata,
            Exception exception);
}

实现该接口的方法，我们可以获取包含发送结果（成功或失败）的异步回调，也就是可以在这个接口的实现中获取发送结果。

我们简单的实现构建一个发布者类，接收主题和发布消息参数，并打印发布结果。

@Component
public class KafkaProducer implements ProducerListener {
    private static final Logger producerlog = LoggerFactory.getLogger(KafkaProducer.class);
    private final KafkaTemplate kafkaTemplate;
    public KafkaProducer(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    public void producer (String msg,String topic){
        ListenableFuture> future = kafkaTemplate.send(topic,0, msg);
        future.addCallback(new KafkaSendCallback() {
            @Override
            public void onSuccess(SendResult result) {
                producerlog.info("发送成功 {}", result);
            }
            @Override
            public void onFailure(KafkaProducerException ex) {
                ProducerRecord failed = ex.getFailedProducerRecord();
                producerlog.info("发送失败 {}",failed);
            }
        });
    }
}

2.1.3 发布消息

写一个controller类来测试我们构建的发布者类，这个类中打印接收到的消息，来确保信息接收不出问题。

@RestController
public class KafkaTestController {
    private static final Logger kafkaTestLog = LoggerFactory.getLogger(KafkaTestController.class);
    @Resource
    private KafkaProducer kafkaProducer;
    @GetMapping("/kafkaTest")
    public void kafkaTest(String msg,String topic){
        kafkaProducer.producer(msg,topic);
        kafkaTestLog.info("接收到消息 {} {}",msg,topic);
    }
}

一切准备就绪，我们启动程序利用postman来进行简单的测试。

进行消息发布：

springboot 连接 kafka集群（kafka版本 2.13-3.4.0）,在这里插入图片描述,第1张

发布结果：

springboot 连接 kafka集群（kafka版本 2.13-3.4.0）,在这里插入图片描述,第2张

可以看到消息发送成功。

我们再看看kafka消费者有没有接收到消息：

springboot 连接 kafka集群（kafka版本 2.13-3.4.0）,在这里插入图片描述,第3张

看以看到，kakfa的消费者也接收到了消息。

2.2 消费者

2.2.1 配置

消息的接受有多种方式，我们这里选择的是使用 @KafkaListener 注解来进行消息接收。它的使用像下面这样：

public class Listener {
    @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
    public void listen(String data) {
        ...
    }
}

看起来不是太难吧，但使用这个注解，我们需要配置底层 ConcurrentMessageListenerContainer.kafkaListenerContainerFactor。

我们在原来的kafka配置类 KafkaConfig 中，继续配置消费者，大概就像下面这样

@Configuration
@EnableKafka
public class KafkaConfig {
     /***** 发布者 *****/
    //生产者工厂
    @Bean
    public ProducerFactory producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }
    //生产者配置
    @Bean
    public Map producerConfigs() {
        Map props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.83:9092,192.168.2.84:9092,192.168.2.86:9092");
           props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }
    //生产者模板
    @Bean
    public KafkaTemplate kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
    /***** 消费者 *****/
    //容器监听工厂
    @Bean
    KafkaListenerContainerFactory>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
    //消费者工厂
    @Bean
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    //消费者配置
    @Bean
    public Map consumerConfigs() {
        Map props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.2.83:9092,192.168.2.84:9092,192.168.2.86:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
        props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);
        return props;
    }
}

注意，要设置容器属性必须使用getContainerProperties()工厂方法。它用作注入容器的实际属性的模板

2.2.2 构建消费者类

配置好后，我们就可以使用这个注解了。这个注解的使用有多种方式：

1、用它来覆盖容器工厂的concurrency和属性

@KafkaListener(id = "myListener", topics = "myTopic",
        autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")
public void listen(String data) {
    ...
}

2、可以使用显式主题和分区（以及可选的初始偏移量）

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(ConsumerRecord record) {
    ...
}

3、将初始偏移应用于所有已分配的分区

@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" },
             partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))
        })
public void listen(ConsumerRecord record) {
    ...
}

4、指定以逗号分隔的分区列表或分区范围

@KafkaListener(id = "pp", autoStartup = "false",
        topicPartitions = @TopicPartition(topic = "topic1",
                partitions = "0-5, 7, 10-15"))
public void process(String in) {
    ...
}

5、可以向侦听器提供Acknowledgment

@KafkaListener(id = "cat", topics = "myTopic",
          containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
    ...
    ack.acknowledge();
}

6、添加标头

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List list,
        @Header(KafkaHeaders.RECEIVED_KEY) List keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List partitions,
        @Header(KafkaHeaders.RECEIVED_TOPIC) List topics,
        @Header(KafkaHeaders.OFFSET) List offsets) {
    ...
}

我们这里写一个简单的，只用它来接受指定主题的数据：

@Component
public class KafkaConsumer {
    private static final Logger consumerlog = LoggerFactory.getLogger(KafkaConsumer.class);
    @KafkaListener(topicPartitions  = @TopicPartition(topic = "kafka-topic-test",
            partitions = "0"))
    public void consumer (String data){
        consumerlog.info("消费者接收数据 {}",data);
    }
}

这里解释一下，因为我们进行了手动分配主题/分区，所以注解中group.id 可以为空。若要指定group.id请在消费者配置中加上props.put(ConsumerConfig.GROUP_ID_CONFIG, “bzt001”); 或在 @TopicPartition 注解后加上 groupId = “组id”

2.2.3 进行消息消费

继续使用postman调用我们写好的发布者发布消息，观察控制台的消费者类是否有相关日志出现。

springboot 连接 kafka集群（kafka版本 2.13-3.4.0）,在这里插入图片描述,第4张

上一篇：解决MySQL8.0报错Client does not support authentication protocol requested by server...问题

下一篇：【Spring Boot】详解restful api