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,
,才可以触发offsets的初始化,我们希望offsets初始化的情况下,能够从kafka进行offsets对比,这种方法解决了初始化的问题。
agent1.channels.c1.zookeeperConnect=10.1.115.181:2181,10.1.114.221:2181,10.1.114.231:2181/kafka
分区调整
但是,还有问题就是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变小问题的主要内容,如果未能解决你的问题,请参考以下文章