标签：reset tank -- auto pros kafka offset latest

kafka auto.offset.reset参数解析

• 1.latest和earliest区别
• 2.创建topic
• 3.生产数据和接收生产数据
• 4.测试代码

auto.offset.reset关乎kafka数据的读取。常用的二个值是latest和earliest，默认是latest。

如果kafka只接收数据，从来没来消费过，程序一开始不要用latest，不然以前的数据就接收不到了。应当先earliest，然后二都都可以。

earliest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
latest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
none
topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常

1.latest和earliest区别

1. earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
2. latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据

提交过offset，latest和earliest没有区别，但是在没有提交offset情况下，用latest直接会导致无法读取旧数据。

2.创建topic

# bin/kafka-topics.sh --create --zookeeper bigserver1:2181,bigserver2:2181,testing:2181 --replication-factor 2 --partitions 3 --topic tank  
Created topic "tank".  
  
# bin/kafka-topics.sh --describe --zookeeper bigserver1:2181,bigserver2:2181,testing:2181 --topic tank  
Topic:tank PartitionCount:3 ReplicationFactor:2 Configs:  
 Topic: tank Partition: 0 Leader: 0 Replicas: 0,2 Isr: 0,2  
 Topic: tank Partition: 1 Leader: 1 Replicas: 1,0 Isr: 1,0  
 Topic: tank Partition: 2 Leader: 2 Replicas: 2,1 Isr: 2,1  

3.生产数据和接收生产数据

[root@bigserver1 kafka]# bin/kafka-console-producer.sh --broker-list bigserver1:9092,bigserver2:9092,testing:9092 --topic tank  
>1  
>2  
>3  
。。。。。。。。。省略。。。。。。。。。  
[root@bigserver1 kafka]# bin/kafka-console-consumer.sh --bootstrap-server bigserver1:9092,bigserver2:9092,testing:9092 --topic tank --from-beginning  
1  
2  
3  

4.测试代码

object tank {  
    def main(args: Array[String]): Unit = {  
        val pros: Properties = new Properties  
        pros.put("bootstrap.servers", "bigserver1:9092,bigserver2:9092,testing:9092")  
        /*分组由消费者决定,完全自定义,没有要求*/  
        pros.put("group.id", "tank")  
        //设置为true 表示offset自动托管到kafka内部的一个特定名称为__consumer_offsets的topic  
        pros.put("enable.auto.commit", "false")  
        pros.put("auto.commit.interval.ms", "1000")  
        pros.put("max.poll.records", "5")  
        pros.put("session.timeout.ms", "30000")  
        //只有当offset不存在的时候，才用latest或者earliest  
        pros.put("auto.offset.reset", "latest")  
  
        pros.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")  
        pros.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")  
  
        val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](pros)  
  
        /*这里填写主题名称*/  
        consumer.subscribe(util.Arrays.asList("tank"))  
  
        val system = akka.actor.ActorSystem("system")  
        system.scheduler.schedule(0 seconds, 30 seconds)(tankTest.saveData(args,consumer))  
  
    }  
  
    object tankTest {  
        def saveData(args: Array[String],consumer: KafkaConsumer[String,String]): Unit = {  
            val records: ConsumerRecords[String, String] = consumer.poll(Duration.ofSeconds(3))  
            if (!records.isEmpty) {  
                for (record <- records) {  
                    if (record.value != null && !record.value.equals("")) {  
                        myLog.syncLog(record.value + "\t准备开启消费者出列数据", "kafka", "get")  
                    }  
                }  
                consumer.commitSync()  
  
            }  
  
        }  
    }  
}  

标签：reset,tank,--,auto,pros,kafka,offset,latest
来源： https://blog.51cto.com/u_15078339/2864054

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。