Spring Boot、kafka、spring-kafka 生产者消费者实践（从搭建kafka集群开始）_auto.create.topics.enable springboot

作者：天景科技苑 | 2024-07-25 10:09:16

踩

auto.create.topics.enable springboot

一、搭建kafka集群

参考文档：http://kafka.apache.org/quickstart 官方文档讲的很详细，而且没坑，照着做很快就可以搭好

注意点 or 建议：

1、在Linux下，启动的kafka集群经常无故退出，看日志也没有报错，就是启动了关闭流程，正常关闭。

解决方案：用守护进程启动,参考：https://blog.csdn.net/xiaoyu_bd/article/details/52268659。

bin/kafka-server-start.sh  -daemon  config/server.properties > k0.log &

2、kafka各项配置以及默认值说明：http://kafka.apache.org/documentation/#configuration 还是官方文档

3、kafka是Java进程，因此可以用 jps 命令方便的查看对应的端口。

4、以下配置为监听地址，默认是 localhost，这样的话无法远程连接，需要配置为特定的IP地址，然后用配置的IP来连接。

listeners=PLAINTEXT://172.17.10.89:9092

如上配置，本地连接也需要使用 172.17.10.89:9092这个地址，而不是 localhost。

二、kafka介绍

参考文档：http://kafka.apache.org/intro 官方文档

https://www.jianshu.com/p/d3e963ff8b70 网友的中文版本，很详细，但要注意有些配置在新版本中发生了变化，比如： auto.create.topics.auto 变成了 auto.create.topics.enable 并且默认值是 true topic不存在时，按照默认配置创建topic

https://www.jianshu.com/p/4e00dff97f39 关闭 offset自动提交，让 spring-kafka 来提交offset

https://blog.csdn.net/lishuangzhe7047/article/details/74530417 kafka auto.offset.reset值详解，offset缺失情况下的消费者消费策略

网上文档很多，不要犹豫该看哪些，而是都看下，对照着看。

三、Spring Boot 集成 kafka

pom spring-kafka 的版本要稍微注意下我用的 kafka_2.11-2.0.0 配合 spring-kafka-1.1.3.RELEASE 可以正常使用，但是用 spring-kafka-2.xxx的时候项目无法启动，会报错：not found


<dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
   <version>1.1.3.RELEASE</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
<dependency>
   <groupId>org.projectlombok</groupId>
   <artifactId>lombok</artifactId>
   <version>1.18.0</version>
   <scope>provided</scope>
</dependency>

yml配置：


spring:
  kafka:
    bootstrap-servers: 172.17.10.89:9092,172.17.10.89:9093,172.17.10.89:9094
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      batch-size: 65536
      buffer-memory: 524288
    consumer:
      group-id: default-group   #默认组id  后面会配置多个消费者组
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: latest  
      enable-auto-commit: false   #关闭自动提交 改由spring-kafka提交
      auto-commit-interval: 100
      max-poll-records: 20      #批量消费 一次接收的最大数量

bootstrap-servers 这边配置的地址可以事先看下是否可以访问到对应的端口

配置读取类：


import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
 
/**
 * @author fandong
 * @create 2018/11/1
 */
@Configuration
@Data
public class KafkaConsumerProps {
 
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
 
    @Value("${spring.kafka.consumer.group-id}")
    private String defaultGroupId;
 
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;
 
    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private String enableAutoCommit;
 
    @Value("${spring.kafka.consumer.auto-commit-interval}")
    private String autoCommitInterval;
 
    @Value("${spring.kafka.consumer.max-poll-records}")
    private String maxPollRecords;
 
    public KafkaConsumerProps() {
    }
}

消费者配置类，配置多个消费者组、批量消费、并发数


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * @author fandong
 * @create 2018/11/1
 */
@Configuration
public class KafkaConsumerConfig {
 
    @Autowired
    private KafkaConsumerProps kafkaConsumerProps;
    
    private static final String GROUP0_ID = "group0";
    private static final String GROUP1_ID = "group1";
 
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory0() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory0());
        //对应topic分区数
        factory.setConcurrency(3);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
 
    public ConsumerFactory<String, String> consumerFactory0() {
        Map<String, Object> map = consumerConfigs();
        map.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP0_ID);
        return new DefaultKafkaConsumerFactory<>(map);
    }
 
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory1() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory1());
        //对应topic分区数
        factory.setConcurrency(3);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }
 
    public ConsumerFactory<String, String> consumerFactory1() {
        Map<String, Object> map = consumerConfigs();
        map.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP1_ID);
        return new DefaultKafkaConsumerFactory<>(map);
    }
 
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>(16);
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProps.getBootstrapServers());
        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerProps.getEnableAutoCommit());
        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerProps.getAutoCommitInterval());
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProps.getDefaultGroupId());
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerProps.getAutoOffsetReset());
        propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerProps.getMaxPollRecords());
        return propsMap;
    }
 
}

注意：

1、factory.setConcurrency(3); 此处设置的目的在于：假设 topic test 下有 0、1、2三个 partition，Spring Boot中只有一个 @KafkaListener() 消费者订阅此 topic，此处设置并发为3，启动后会有三个不同的消费者分别订阅 p0、p1、p2，本地实际有三个消费者线程。而 factory.setConcurrency(1); 的话本地只有一个消费者线程， p0、p1、p2被同一个消费者订阅。由于一个partition只能被同一个消费者组下的一个消费者订阅，对于只有一个 partition的topic，即使设置并发为3，也只会有一个消费者，多余的消费者没有 partition可以订阅。

2、factory.setBatchListener(true); 设置批量消费，每个批次数量在Kafka配置参数ConsumerConfig.MAX_POLL_RECORDS_CONFIG中配置，限制的是一次批量接收的最大条数，而不是等到达到最大条数才接收，这点容易被误解。实际测试时，接收是实时的，当生产者大量写入时，一次批量接收的消息数量为配置的最大条数。

生产者我们借助自动配置，在yml文件中加入生产者配置之后，直接注入 KafkaTemplate 即可使用。


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
 
import java.util.List;
 
/**
 * @author fandong
 * @create 2018/11/1
 */
@Service
public class KafkaServiceImpl implements KafkaService {
 
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
 
    @Autowired
    public KafkaServiceImpl(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = (KafkaTemplate<String, String>) kafkaTemplate;
    }
 
    @Override
    public void send(String topic, String value) {
        ListenableFuture<SendResult<String, String>> resultListenableFuture = kafkaTemplate.send(topic, value);
        resultListenableFuture.addCallback(
                successCallback -> logger.info("发送成功：topic= " + topic + " value= " + value),
                failureCallback -> logger.info("发送失败：topic= " + topic + " value= " + value));
    }
 
    @Override
    @KafkaListener(topics = {"test"}, containerFactory = "kafkaListenerContainerFactory0")
    public void kafkaConsumerTest(String msg) {
        logger.info("接收到消息--" + msg);
    }
 
    @Override
    @KafkaListener(topics = {"3-test"}, containerFactory = "kafkaListenerContainerFactory0")
    public void listenPartition0(List<ConsumerRecord<String, String>> records) {
        System.out.println(records.size());
        for (ConsumerRecord<String, String> consumerRecord : records){
            String value = consumerRecord.value();
            logger.info("a 消息：partition " + consumerRecord.partition() + " value " + consumerRecord.value() );
        }
    }
 
    @Override
    @KafkaListener(topics = {"3-test"}, containerFactory = "kafkaListenerContainerFactory1")
    public void listenPartition2(List<ConsumerRecord<String, String>> records) {
        System.out.println(records.size());
        for (ConsumerRecord<String, String> consumerRecord : records){
            String value = consumerRecord.value();
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            logger.info("c 消息：partition " + consumerRecord.partition() + " value " + consumerRecord.value() + " thread id " + Thread.currentThread().getName());
        }
    }
 
 
}

指定 containerFactory = "kafkaListenerContainerFactory1" 参数给消费者分组，值为之前定义的 KafkaListenerContainerFactory的 Bean 名称，不指定的情况下默认是方法名称。

注：topic 3-test下有 3个partition，由于之前配置了 factory.setConcurrency(3); 项目启动之后，本地会有三个消费者线程。

使用如下命令可以查看各个消费者组的情况以下为查看 group0消费组，可以看到每个partition由不同的消费者订阅。


bin/kafka-consumer-groups.sh --bootstrap-server 172.17.10.89:9092 --describe --group group0
 
TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                     HOST            CLIENT-ID
test            0          12              12              0               consumer-1-9419e037-1501-4f33-85c9-25df75e3a5a9 /172.17.10.33   consumer-1
3-test          1          1400            1400            0               consumer-8-1a520c8e-c412-4178-a76f-77c68e7472b7 /172.17.10.33   consumer-8
3-test          0          1398            1398            0               consumer-7-ede99b63-ea10-48dd-be62-61139360e39c /172.17.10.33   consumer-7
3-test          2          1397            1397            0               consumer-9-4bacc023-7de7-4e98-aefd-bb39d0bf6547 /172.17.10.33   consumer-9

四、关于提高消费者消费能力的思考

kafka写具有很好的性能，而消费者在消费时往往会有相对耗时的操作，所以经常出现消费者性能跟不上的情况。

思路：

1、在topic下适当创建多个 partition，然后使用多个消费者来消费多个partition

2、使用批量消费，一次接收多条消息，相比一个一个接收，（猜）可以减少IO次数，提高速度

3、消费者再使用线程池配合适当长度的阻塞队列，进一步提高处理能力（需要分析任务类型以及考虑处理器的能力）。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/天景科技苑/article/detail/879681