行业知识
kafka文件存储机制和消费者
2026-11-06 01:22:26
来源：互联网转载
1.broker文件存储机制
去���看真正的存储文件：
在/opt/module/kafka/datas/ 路径下
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
如果是6415那么这个会存储在563的log文件之中，因为介于6410和10090之间。
2.文件清除策略
日志压缩用的比较少，比如针对用户，18岁19岁，只需要保存最新的就行。
3.高效读写数据
4.kafka消费者
5.消费者组初始化流程
6.独立消费者案例
package com.atguigu.kafka.producer;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.clients.producer.ProducerRecord;import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;public class CustomProducer { public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop100:9092"); // key,value 序列化（必须）：key.serializer，value.serializer properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); // 3. 创建 kafka 生产者对象 KafkaProducer kafkaProducer = new KafkaProducer(properties); // 4. 调用 send 方法,发送消息 for (int i = 0; i 7.独立消费者案例（订阅分区）
package com.atguigu.kafka.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.ArrayList;import java.util.Properties;public class CustomConsumerPartition { public static void main(String[] args) { // 1. 创建 kafka 生产者的配置对象 Properties properties = new Properties(); // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers Object put = properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop100:9092,hadoop101:9092"); properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test"); // 3. 创建 kafka 生产者对象 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); ArrayList topicPartition = new ArrayList(); topicPartition.add(new TopicPartition("first",0)); kafkaConsumer.assign(topicPartition); while (true){ ConsumerRecords records = kafkaConsumer.poll(Duration.ofSeconds(1)); for(ConsumerRecord s : records) { System.out.println(s); } } }} 8.分区分配以及再平衡策略
1、一个consumer group中有多个consumer组成，一个 topic有多个partition组成，现在的问题是，到底由哪个consumer来消费哪个 partition的数据。
2、Kafka有四种主流的分区分配策略： Range、RoundRobin、Sticky、CooperativeSticky。 可以通过配置参数partition.assignment.strategy，修改分区的分配策略。默认策略是Range + CooperativeSticky。Kafka可以同时使用 多个分区分配策略。
// 修改分区分配策略 properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFI G, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
// 修改分区分配策略properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.StickyAssignor");
9. offset 的默认维护位置
consumer_offsets 主题里面采用 key 和 value 的方式存储数据。key 是 group.id+topic+ 分区号，value 就是当前 offset 的值。每隔一段时间，kafka 内部会对这个 topic 进行 compact，也就是每个 group.id+topic+分区号就保留最新数据。
10.自动提交 offset
// 是否自动提交 offset properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true); // 提交 offset 的时间周期 1000ms，默认 5sproperties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);
11.手动提交offset
虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此 吞吐量会受到很大的影响。因此更多的情况下，会选用异步提交 offset 的方式。
12.指定offset消费
package com.atguigu.kafka.consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.ArrayList;import java.util.Properties;import java.util.Set;public class CustomConsumerSeek { public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092"); // key value 反序列化 properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2"); // 1 创建一个消费者 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); // 2 订阅一个主题 ArrayList topics = new ArrayList(); topics.add("first"); kafkaConsumer.subscribe(topics); Set assignment = kafkaConsumer.assignment(); //确保分区分配方案已经制定完毕 while (assignment.size() == 0){ kafkaConsumer.poll(Duration.ofSeconds(1)); assignment = kafkaConsumer.assignment(); } for(TopicPartition topicPartition : assignment) { kafkaConsumer.seek(topicPartition,100); } while (true) { ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } }} 13.指定时间消费
需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。 例如要求按照时间消费前一天的数据，怎么处理？
package com.atguigu.kafka.consumer;import org.apache.kafka.clients.consumer.*;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.*;public class CustomConsumerSeekTime { public static void main(String[] args) { // 0 配置信息 Properties properties = new Properties(); // 连接 properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092"); // key value 反序列化 properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2"); // 1 创建一个消费者 KafkaConsumer kafkaConsumer = new KafkaConsumer(properties); // 2 订阅一个主题 ArrayList topics = new ArrayList(); topics.add("first"); kafkaConsumer.subscribe(topics); Set assignment = kafkaConsumer.assignment(); //确保分区分配方案已经制定完毕 while (assignment.size() == 0){ kafkaConsumer.poll(Duration.ofSeconds(1)); assignment = kafkaConsumer.assignment(); } HashMap objectObjectHashMap = new HashMap(); for (TopicPartition topicPartition : assignment) { objectObjectHashMap.put(topicPartition,System.currentTimeMillis()-1*24*3600*1000); } Map topicPartitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(objectObjectHashMap); for(TopicPartition topicPartition : assignment) { OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition); kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset()); } while (true) { ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord consumerRecord : consumerRecords) { System.out.println(consumerRecord); } } }}
