flume高并发优化——(16)解决offsets变小问题

Posted ESOO

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了flume高并发优化——(16)解决offsets变小问题相关的知识,希望对你有一定的参考价值。

offsets初始化

在上篇博客中《flume高并发优化——(14)解决空行停止收集数据问题,及offsets变小问题 》我们遗留了一个小问题,就是offsets变小的问题,迟迟未解决,经过研究flume代码发现,flume中,是自己管理offsets关系的,每个kafkachannel的代码中保留了一份topic-offsets的关系,源码:

@Override
  public void start() 
    try 
      logger.info("=KafkaChannel.start=> begin " + getName());
      // As a migration step check if there are any offsets from the group stored in kafka
      // If not read them from Zookeeper and commit them to Kafka
      //***重要:设置这两个属性才可以初始化offsets***
      if (migrateZookeeperOffsets && zookeeperConnect != null && !zookeeperConnect.isEmpty()) 
        migrateOffsets();
      
      logger.info("=KafkaChannel.start=> create KafkaProducer begin producerProps:" + JSON.toString(producerProps));
      producer = new KafkaProducer<String, byte[]>(producerProps);
      // We always have just one topic being read by one thread
      logger.info("Topic = ", topic.get());
      counter.start();
      super.start();
    catch (Exception ex)
      logger.error("=KafkaChannel.start=>error:",ex);
    
    logger.info("=KafkaChannel.start=> end " + getName());

  

我们发现,设置了agent1.channels.c1.migrateZookeeperOffsets=true,
agent1.channels.c1.zookeeperConnect=10.1.115.181:2181,10.1.114.221:2181,10.1.114.231:2181/kafka
,才可以触发offsets的初始化,我们希望offsets初始化的情况下,能够从kafka进行offsets对比,这种方法解决了初始化的问题。

分区调整

但是,还有问题就是offsets的变化问题,研究发现,在offsets提交中,遇到消费者多余分区的情况,在某些消费者重新分配的情况下,因为每个消费者都保留了topic-offsets对应关系,导致offsets消费重复,解决方法是改为自动提交,且分区>=消费者,修正后源码:

private void setConsumerProps(Context ctx, String bootStrapServers) 
    consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, DEFAULT_KEY_DESERIALIZER);
    consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DEFAULT_VALUE_DESERIAIZER);
    consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, DEFAULT_AUTO_OFFSET_RESET);
    //Defaults overridden based on config
    consumerProps.putAll(ctx.getSubProperties(KAFKA_CONSUMER_PREFIX));
    //These always take precedence over config
    consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
    consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    //***重要:设置这两个属性才自动提交offsets***
    consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);  
    //***重要:自动提交offsets频率 ms ***
    consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
  

总结:

遇到此类问题,还是我们对于kafka的设计不熟悉,导致的问题,在源码的层次分析问题帮助我们更容易找到问题的梗结所在,推荐各大公司使用开源项目,便于定制与维护。

以上是关于flume高并发优化——(16)解决offsets变小问题的主要内容,如果未能解决你的问题,请参考以下文章

flume高并发优化——(15)中间件版本升级

flume高并发优化——多文件source扩展断点续传

flume高并发优化——(13)扩展三级文件配置&利用Headers扩展属性

Web性能优化 高并发网站解决 单例

HttpClient对高并发有啥优化吗

大流量高并发量网站的之解决方案