Kafka Api和message、segment结构介绍

1. segment文件

一个partition当中由多个segment文件组成,每个segment文件,包含两部分,一个是 .log 文件,另外一个是 .index 文件,其中 .log 文件包含了我们发送的数据存储,.index 文件,记录的是我们.log文件的数据索引值,以便于我们加快数据的查询速度。

.index 与 .log 对应关系如下:
Kafka Api和message、segment结构介绍_第1张图片
上图左半部分是索引文件,里面存储的是一对一对的key-value。

key是消息在数据文件(对应的log文件)中的编号,比如“1,3,6,8……”, 分别表示在log文件中的第1条消息、第3条消息、第6条消息、第8条消息……

那么为什么在index文件中这些编号不是连续的呢?这是因为index文件中并没有为数据文件中的每条消息都建立索引,而是采用了稀疏存储的方式,每隔一定字节的数据建立一条索引。这样避免了索引文件占用过多的空间,从而可以将索引文件保留在内存中。但缺点是没有建立索引的Message也不能一次定位到其在数据文件的位置,从而需要做一次顺序扫描,但是这次顺序扫描的范围就很小了。

value 代表的是在全局partiton中的第几个消息。

以索引文件中元数据 3,497 为例,其中3代表在右边log数据文件中从上到下第3个消息, 497表示该消息的物理偏移地址(位置)为497(也表示在全局partiton表示第497个消息-顺序写入特性)。

log日志目录及组成kafka在我们指定的log.dir目录下,会创建一些文件夹;名字是 (主题名字-分区名) 所组成的文件夹。在(主题名字-分区名)的目录下,会有两个文件存在,如下所示:

#索引文件
00000000000000000000.index
#日志内容
00000000000000000000.log

在目录下的文件,会根据log日志的大小进行切分,.log文件的大小为1G的时候,就会进行切分文件;如下:

-rw-r--r--. 1 root root 389k  117  18:03   00000000000000000000.index
-rw-r--r--. 1 root root 1.0G  117  18:03   00000000000000000000.log
-rw-r--r--. 1 root root  10M  117  18:03   00000000000000077894.index
-rw-r--r--. 1 root root 127M  117  18:03   00000000000000077894.log

segment文件命名规则:partion全局的第一个segment从0开始,后续每个segment文件名为上一个全局 partion的最大offset(偏移message数)。数值最大为64位long大小,20位数字字符长度,没有数字就用 0 填充。

2.kafka message的物理结构

生产者发送到kafka的每条消息,都被kafka包装成了一个message

message 的物理结构如下图所示:

Kafka Api和message、segment结构介绍_第2张图片
所以生产者发送给kafka的消息并不是直接存储起来,而是经过kafka的包装,每条消息都是上图这个结构,只有最后一个字段才是真正生产者发送的消息数据。

3.1. 生产者代码

/**
* 订单的生产者代码,
*/
public class OrderProducer {
     
public static void main(String[] args) throws InterruptedException {
     

Properties props = new Properties(); props.put("bootstrap.servers", "node01:9092"); props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432); 
props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer"); 
props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
 KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>
(props);
for (int i = 0; i < 1000; i++) {
     
// 发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value 
kafkaProducer.send(new ProducerRecord<String, String>("order", "订单信
息!"+i));
Thread.sleep(100);
}
}
}

kafka当中支持以下四种数据的分区方式:

//第一种分区策略,如果既没有指定分区号,也没有指定数据key,那么就会使用轮询的方式将数据均匀的发送到不同的分区里面去
  //ProducerRecord producerRecord1 = new ProducerRecord<>("mypartition", "mymessage" + i);
  //kafkaProducer.send(producerRecord1);
  //第二种分区策略 如果没有指定分区号,指定了数据key,通过key.hashCode  % numPartitions来计算数据究竟会保存在哪一个分区里面
  //注意:如果数据key,没有变化   key.hashCode % numPartitions  =  固定值  所有的数据都会写入到某一个分区里面去
  //ProducerRecord producerRecord2 = new ProducerRecord<>("mypartition", "mykey", "mymessage" + i);
  //kafkaProducer.send(producerRecord2);
  //第三种分区策略:如果指定了分区号,那么就会将数据直接写入到对应的分区里面去
//  ProducerRecord producerRecord3 = new ProducerRecord<>("mypartition", 0, "mykey", "mymessage" + i);
 // kafkaProducer.send(producerRecord3);
  //第四种分区策略:自定义分区策略。如果不自定义分区规则,那么会将数据使用轮询的方式均匀的发送到各个分区里面去
  kafkaProducer.send(new ProducerRecord<String, String>("mypartition","mymessage"+i));

自定义分区策略:

public class KafkaCustomPartitioner implements Partitioner {
     
 @Override
 public void configure(Map<String, ?> configs) {
     
 }

 @Override
 public int partition(String topic, Object arg1, byte[] keyBytes, Object arg3, byte[] arg4, Cluster cluster) {
     
  List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
     int partitionNum = partitions.size();
  Random random = new Random();
  int partition = random.nextInt(partitionNum);
     return partition;
 }

 @Override
 public void close() {
     
  
 }

}

主代码中添加配置:

public void kafkaProducer() throws Exception {
     
  //1、准备配置文件
     Properties props = new Properties();
     props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
     props.put("acks", "all");
     props.put("retries", 0);
     props.put("batch.size", 16384);
     props.put("linger.ms", 1);
     props.put("buffer.memory", 33554432);
     props.put("partitioner.class", "cn.itcast.kafka.partitioner.KafkaCustomPartitioner");
     props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
     props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
     //2、创建KafkaProducer
     KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
     for (int i=0;i<100;i++){
     
         //3、发送数据
         kafkaProducer.send(new ProducerRecord<String, String>("testpart","0","value"+i));
     }

  kafkaProducer.close();
 }

3.2 消费者代码

1) 自动提交offset

/**
* 消费订单数据--- javaben.tojson
*/
public class OrderConsumer {
     
public static void main(String[] args) {
     
// 1\连接集群
Properties props = new Properties(); props.put("bootstrap.servers", "hadoop-01:9092"); props.put("group.id", "test");

//以下两行代码 ---消费者自动提交offset值 
props.put("enable.auto.commit", "true"); 
props.put("auto.commit.interval.ms",  "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>
(props);
//   2、发送数据 发送数据需要,订阅下要消费的topic。order kafkaConsumer.subscribe(Arrays.asList("order")); 
while (true) {
     
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);// jdk queue offer插入、poll获取元素。blockingqueue put插入原生, take获取元素
for (ConsumerRecord<String, String> record : consumerRecords) {
      System.out.println("消费的数据为:" + record.value());
}
}
}
}

2) 手动提交offset
如果Consumer在获取数据后,需要加入处理,数据完毕后才确认offset,需要程序来控制offset的确认。

关闭自动提交确认选项:props.put(“enable.auto.commit”, “false”);

手动提交offset值:kafkaConsumer.commitSync();

Properties props = new Properties(); 
props.put("bootstrap.servers", "localhost:9092"); 
props.put("group.id", "test");
//关闭自动提交确认选项
props.put("enable.auto.commit", "false"); 
props.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); 
props.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("test"));
final int minBatchSize = 200;
List<ConsumerRecord<String, String>> buffer = new ArrayList<>(); 
while (true) {
     
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
     
buffer.add(record);
}
if (buffer.size() >= minBatchSize) {
      
insertIntoDb(buffer);
// 手动提交offset值
consumer.commitSync(); 
buffer.clear();
}
}

3) 消费完每个分区之后手动提交offset
上面的示例使用commitSync将所有已接收的记录标记为已提交。在某些情况下,可能希望通过明确指定偏移量来更好地控制已提交的记录。在下面的示例中,我们在完成处理每个分区中的记录后提交偏移量:

try {
     
while(running) {
     
ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE); 
for (TopicPartition partition : records.partitions()) {
     
List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
for (ConsumerRecord<String, String> record : partitionRecords) {
      System.out.println(record.offset() + ": " + record.value());
}
long lastOffset = partitionRecords.get(partitionRecords.size() -1).offset();
consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
}
}
} finally {
      consumer.close();}

注意事项:

提交的偏移量应始终是应用程序将读取的下一条消息的偏移量。因此,在调用commitSync(偏移量)时,应该在最后处理的消息的偏移量中添加一个。

4) 指定分区数据进行消费
如果进程正在维护与该分区关联的某种本地状态(如本地磁盘上的键值存储),那么它应该只获取它在磁盘上维护的分区的记录。

如果进程本身具有高可用性,并且如果失败则将重新启动(可能使用YARN,Mesos或AWS工具等集群管理框 架,或作为流处理框架的一部分)。在这种情况下,Kafka不需要检测故障并重新分配分区,因为消耗过程将在另一台机器上重新启动。

Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test"); 
props.put("enable.auto.commit", "true");
 props.put("auto.commit.interval.ms", "1000"); 
props.put("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); 
props.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer"); 
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//consumer.subscribe(Arrays.asList("foo",  "bar"));

//手动指定消费指定分区的数据---start 
String topic = "foo";
TopicPartition partition0 = new TopicPartition(topic, 0); 
TopicPartition partition1 = new TopicPartition(topic, 1); consumer.assign(Arrays.asList(partition0,  partition1));
//手动指定消费指定分区的数据---end
while (true) {
     
ConsumerRecords<String, String> records = consumer.poll(100); 
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

你可能感兴趣的