devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka优化(系列四)：kafka配置优化和kafka批量消费，提高分区数量_o.s.kafka.listener.kafkamessagelistenercontainer -

作者：我家自动化 | 2024-02-21 17:54:08

赞

踩

o.s.kafka.listener.kafkamessagelistenercontainer - partitions assigned

往期精选

    这一篇我们将叙述，我是怎么在项目中进行对kafka优化的我们将从三方面进行考虑，一是代码；二是    配置；三是集群。项目背景，做数据迁移工作后面我将写几篇文章讲诉我们是怎么对百万数据进行迁移的工作）。主要场景利用kafka做读写分离，一直请求源数据写入到kafka生产者，然后kafka消费者进行写入数据到新数据。

一、配置优化《报错》节选：

  1. [2018-09-25 11:23:59.370] ERROR [org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1] LoggingErrorHandler.java:37 - Error while processing: null
  2.  
  3. org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.
  4.  
  5. at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:722)
  6.  
  7. at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:600)
  8.  
  9. at org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1250)
  10.  
  11. at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.commitIfNecessary(KafkaMessageListenerContainer.java:1324)
  12.  
  13. at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.processCommits(KafkaMessageListenerContainer.java:1185)
  14.  
  15. at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:688)
  16.  
  17. at java.util.concurrent.Executors$RunnableAdapter.call$$$capture(Executors.java:511)
  18.  
  19. at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java)
  20.  
  21. at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
  22.  
  23. at java.util.concurrent.FutureTask.run(FutureTask.java)
  24.  
  25. at java.lang.Thread.run(Thread.java:748)

这是由于kafka一直生产数据，导致kafka消费太慢了。我们主要优化也是对消费者进行优化。根据上面的报错，我们可以看到一个参数：max-poll-records，所以我们首先将对提交数，进行调大。具体的需要根据项目进行测试，我们把数进行调大到100，同时对下面的参数进行：

#自动提交offset到zookeeper的时间间隔

auto-commit-interval: 1000

#earliest 

#当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 

#latest 

#当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 

#none 

#topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

auto-offset-reset: latest

#提交方式改为false，是否自动周期性提交已经拉取到消费端的消息offset

enable-auto-commit: false

由于使用了spring-kafka，则把kafka-client的enable.auto.commit设置成了false，表示禁止kafka-client自动提交offset，因为就是之前的自动提交失败，导致offset永远没更新，从而转向使用spring-kafka的offset提交机制。并且spring-kafka提供了多种提交策略：

  然后我修改了kafka的配置(spring-kafka)，需要到安装的文件(config)下进行修改，分别是生产文件和配置文件。

1.session.timeout.ms=100000(增大session超时时间)。

2.request.timeout.ms=110000(socket握手超时时间,默认是3000 但是kafka配置要求大于session.timeout.ms时间).

同时Kafka的生产端可以压缩消息，如果原始消息是XML，当通过压缩之后，消息可能会变得不那么大。在生产端的配置参数中使用compression.codec和commpressed.topics可以开启压缩功能，压缩算法可以使用GZip或Snappy。

二、代码优化：《日志》节选

如果进行了的配置调优，差不多会提高kafka的消费能力，但是写入过大，控制台还是打印下面日志信息：

  1. 2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] LogContext.java:341 - [Consumer clientId=consumer-4, groupId=test-consumer-group] Revoking previously assigned partitions [XXXXX-0]
  2.  
  3. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] LogContext.java:341 - [Consumer clientId=consumer-1, groupId=test-consumer-group] Revoking previously assigned partitions [XXXXX-0]
  4.  
  5. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#4-0-C-1] LogContext.java:341 - [Consumer clientId=consumer-3, groupId=test-consumer-group] Revoking previously assigned partitions [XXXXXX-0]
  6.  
  7. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] AbstractMessageListenerContainer.java:343 - partitions revoked: [XXXX-0]
  8.  
  9. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#2-0-C-1] AbstractMessageListenerContainer.java:343 - partitions revoked: [XXXXXX-0]
  10.  
  11. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#4-0-C-1] AbstractMessageListenerContainer.java:343 - partitions revoked: [XXXXXX-0]
  12.  
  13. [2018-09-25 14:39:53.193] INFO [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] LogContext.java:336 - [Consumer clientId=consumer-4, groupId=test-consumer-group] 

也是就说机制一直打印这些信息，但是又不报错，但是又不写入数据，我们就想，除了配置优化之后，能不能像数据库一样，批量提交或者说是批量消费呢？看了官网资料，发现确实可以，以下是我们对代码的优化，由单一的消费，改为批量消费：

一：增加一个config类。

  1. @Configuration
  2. @EnableKafka
  3.  
  4. public class KafkaConfig {
  5.  
  6.  
  7.  
  8.     @Bean
  9.  
  10.     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
  12.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  13.  
  14.         factory.setConsumerFactory(consumerFactory());
  15.  
  16.         factory.setConcurrency(10);
  17.  
  18.         factory.getContainerProperties().setPollTimeout(1500);
  19.  
  20.         factory.setBatchListener(true);
  21.  
  22.         //@KafkaListener 批量消费  每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
  23.  
  24.         factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
  25.  
  26.         //设置提交偏移量的方式
  27.  
  28.         return factory;
  29.  
  30.     }
  31.  
  32.  
  33.  
  34.     public ConsumerFactory<String, String> consumerFactory() {
  35.  
  36.         return new DefaultKafkaConsumerFactory<>(consumerConfigs());
  37.  
  38.     }
  39.  
  40.     public Map<String, Object> consumerConfigs() {
  41.  
  42.         Map<String, Object> propsMap = new HashMap<>(16);
  43.  
  44.         propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP地址需要修改");
  45.  
  46.         propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  47.  
  48.         propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
  49.  
  50.         propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 100000);
  51.  
  52.         propsMap.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,110000);
  53.  
  54.         propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  55.  
  56.         propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  57.  
  58.         propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
  59.  
  60.         propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  61.  
  62.         propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 150);
  63.  
  64.         //每个批次获取数
  65.  
  66.         return propsMap;
  67.  
  68.     }
  69.  
  70.  
  71.  
  72. }

二、更改消费接受代码。

  1.  @KafkaListener(topics = {"消费名称需要改"})
  2.  
  3.     public void listen(List<ConsumerRecord> records, Acknowledgment ack) {
  4.  
  5.         try {
  6.  
  7.             for (ConsumerRecord record : records) {
  8.  
  9.                 Optional<?> kafkaMessage = Optional.ofNullable(record.value());
  10.  
  11.                 if (kafkaMessage.isPresent()) {
  12.  
  13.                     Object message = kafkaMessage.get();
  14.  
  15.                     log.info("----------------- record =" + record);
  16.  
  17.                     log.info("------------------ message =" + message);
  18.  
  19.                 }
  20.  
  21.             }
  22.  
  23.         } catch (Exception e) {
  24.  
  25.             log.error("kafka失败，当前失败的批次。data:{}", records);
  26.  
  27.             e.printStackTrace();
  28.  
  29.         } finally {
  30.  
  31.             ack.acknowledge();
  32.  
  33.         }
  34.  
  35.  
  36.  
  37.     }

集群搭建

前面虽然优化配置和代码，但是代码执行还是不够快，网上寻找资料（提高了partition的数量，从而提高了consumer的并行能力，从而提高数据的消费能力），说可以提高分区数量，如果单机怎么提高还是一样的(我们试过了)，后来搭建了一个集群。注意我们是使用docker搭建kafka集群的，搭建过程如下。docker-compose.yml内容：

  1. version: '2'
  2.  
  3. services:
  4.  
  5.   zookeeper:
  6.  
  7.     image: wurstmeister/zookeeper
  8.  
  9.     ports:
  10.  
  11.       - "2181:2181"
  12.  
  13.   kafka:
  14.  
  15.     image: wurstmeister/kafka
  16.  
  17.     ports:
  18.  
  19.       - "9095:9095"
  20.  
  21.     environment:
  22.  
  23.       KAFKA_ADVERTISED_HOST_NAME: IP地址
  24.  
  25.       KAFKA_ADVERTISED_PORT: 9095
  26.  
  27.       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://XXXXXXXX:9095
  28.  
  29.       KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9095
  30.  
  31.       KAFKA_DELETE_TOPIC_ENABLE: "true"
  32.  
  33.       KAFKA_LOG_RETENTION_HOURS: 1
  34.  
  35.       KAFKA_MESSAGE_MAX_BYTES: 10000000
  36.  
  37.       KAFKA_REPLICA_FETCH_MAX_BYTES: 10000000
  38.  
  39.       KAFKA_GROUP_MAX_SESSION_TIMEOUT_MS: 100000
  40.  
  41.       KAFKA_NUM_PARTITIONS: 2
  42.  
  43.       KAFKA_DELETE_RETENTION_MS: 1000
  44.  
  45.       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
  46.  
  47.       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  48.  
  49.     volumes:
  50.  
  51.       - /var/run/docker.sock:/var/run/docker.sock
  52.  
  53.  
  54.  
  55.   kafka-manager:
  56.  
  57.     image: sheepkiller/kafka-manager
  58.  
  59.     links:
  60.  
  61.       - kafka
  62.  
  63.       - zookeeper
  64.  
  65.     environment:
  66.  
  67.       ZK_HOSTS: zookeeper:2181
  68.  
  69.       APPLICATION_SECRET: letmein
  70.  
  71.       KM_ARGS: -Djava.net.preferIPv4Stack=true
  72.  
  73.     ports:
  74.  
  75.       - "9000:9000"

1.启动的命令：

docker-compose up -d

2.先去修改配置文件的端口，然后再启动相关的命令：

docker-compose scale kafka=2

3.再次修改文件袋的端口，然后再启动相关的命令：

docker-compose scale kafka=3

以上就是我所总结的kafka优化，欢迎有更好的方案进行交流，欢迎关注微信号：繁荣Aaron和转发。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/125502
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号