【Kafka笔记】Kafka API详细解析 Java版本（Producer API，Consumer API，拦截器等）

根据send方法的参数的构造方法就可以看出来，

指定分区就发送到指定分区

没有指定分区，有key值，就按照key值的Hash值分配分区

没有指定分区，也没有指定key值，轮询分区分配（只分配一次，以后都按照第一次的分区顺序）

自定义分区器需要实现org.apache.kafka.clients.producer.Partitioner接口。并且实现三个方法

public class KafkaMyPartitions implements Partitioner { @Override public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) { return 0; } @Override public void close() { } @Override public void configure(Map map) { } }

自定义分区实例：

KafkaMyPartitions：

public class KafkaMyPartitions implements Partitioner { @Override public int partition(String s, Object o, byte[] bytes, Object o1, byte[] bytes1, Cluster cluster) { // 这里写自己的分区策略 // 我这里指定为1 return 1; } @Override public void close() { } @Override public void configure(Map map) { } }

KafkaProducerCallbackDemo：

public class KafkaProducerCallbackDemo { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "XXXXX:9093");//kafka集群，broker-list props.put("acks", "all"); props.put("retries", 1);//重试次数 props.put("batch.size", 16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 指定自定义分区 props.put("partitioner.class","com.firehome.newkafka.KafkaMyPartitions"); // 创建KafkaProducer客户端 KafkaProducer producer = new KafkaProducer<>(props); for (int i = 20; i < 25 ; i++) { producer.send(new ProducerRecord("th-topic", "ImKey-" + i, "ImValue-" + i), new Callback() { @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (e == null){ System.out.printf("消息发送成功！topic=%s,partition=%s,offset=%d \n",recordMetadata.topic(),recordMetadata.partition(),recordMetadata.offset()); }else { System.err.println("消息发送失败！"); } } }); } producer.close(); } }

返回日志：

消息发送成功！topic=th-topic,partition=1,offset=27 消息发送成功！topic=th-topic,partition=1,offset=28 消息发送成功！topic=th-topic,partition=1,offset=29 消息发送成功！topic=th-topic,partition=1,offset=30 消息发送成功！topic=th-topic,partition=1,offset=31

可以看到直接发送到了分区1上了。

Producer API是线程安全的，直接就可以使用多线程发送消息，实例：

public class KafkaProducerThread implements Runnable { private KafkaProducer kafkaProducer; public KafkaProducerThread(){ } public KafkaProducerThread(KafkaProducer kafkaProducer){ this.kafkaProducer = kafkaProducer; } @Override public void run() { for (int i = 0; i < 20 ; i++) { String key = "ImKey-" + i+"-"+Thread.currentThread().getName(); String value = "ImValue-" + i+"-"+Thread.currentThread().getName(); kafkaProducer.send(new ProducerRecord<>("th-topic", key, value)); System.out.printf("Thread-name = %s, key = %s, value = %s",Thread.currentThread().getName(),key,value); } } public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "XXXXXXXX:9093");//kafka集群，broker-list props.put("acks", "all"); props.put("retries", 1);//重试次数 props.put("batch.size", 16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 创建KafkaProducer客户端 KafkaProducer producer = new KafkaProducer<>(props); KafkaProducerThread producerThread1 = new KafkaProducerThread(producer); //KafkaProducerThread producerThread2 = new KafkaProducerThread(producer); Thread one = new Thread(producerThread1, "one"); Thread two = new Thread(producerThread1, "two"); System.out.println("线程开始"); one.start(); two.start(); } }

这里只是一个简单的实例。

kafka客户端通过TCP长连接从集群中消费消息，并透明地处理kafka集群中出现故障服务器，透明地调节适应集群中变化的数据分区。也和服务器交互，平衡均衡消费者。

kafka为分区中的每条消息保存一个偏移量（offset），这个偏移量是该分区中一条消息的唯一标示符。也表示消费者在分区的位置。例如，一个位置是5的消费者(说明已经消费了0到4的消息)，下一个接收消息的偏移量为5的消息。实际上有两个与消费者相关的“位置”概念：

消费者的位置给出了下一条记录的偏移量。它比消费者在该分区中看到的最大偏移量要大一个。 它在每次消费者在调用poll(long)中接收消息时自动增长。

“已提交”的位置是已安全保存的最后偏移量，如果进程失败或重新启动时，消费者将恢复到这个偏移量。消费者可以选择定期自动提交偏移量，也可以选择通过调用commit API来手动的控制(如：commitSync 和 commitAsync)。

这个区别是消费者来控制一条消息什么时候才被认为是已被消费的，控制权在消费者。

Kafka的消费者组概念，通过进程池瓜分消息并处理消息。这些进程可以在同一台机器运行，也可分布到多台机器上，以增加可扩展性和容错性，相同group.id的消费者将视为同一个消费者组。

分组中的每个消费者都通过subscribe API动态的订阅一个topic列表。kafka将已订阅topic的消息发送到每个消费者组中。并通过平衡分区在消费者分组中所有成员之间来达到平均。因此每个分区恰好地分配1个消费者（一个消费者组中）。所有如果一个topic有4个分区，并且一个消费者分组有只有2个消费者。那么每个消费者将消费2个分区。

消费者组的成员是动态维护的：如果一个消费者故障。分配给它的分区将重新分配给同一个分组中其他的消费者。同样的，如果一个新的消费者加入到分组，将从现有消费者中移一个给它。这被称为重新平衡分组。当新分区添加到订阅的topic时，或者当创建与订阅的正则表达式匹配的新topic时，也将重新平衡。将通过定时刷新自动发现新的分区，并将其分配给分组的成员。

从概念上讲，你可以将消费者分组看作是由多个进程组成的单一逻辑订阅者。作为一个多订阅系统，Kafka支持对于给定topic任何数量的消费者组，而不重复。

这是在消息系统中常见的功能的略微概括。所有进程都将是单个消费者分组的一部分（类似传统消息传递系统中的队列的语义），因此消息传递就像队列一样，在组中平衡。与传统的消息系统不同的是，虽然，你可以有多个这样的组。但每个进程都有自己的消费者组（类似于传统消息系统中pub-sub的语义），因此每个进程都会订阅到该主题的所有消息。

此外，当分组重新分配自动发生时，可以通过ConsumerRebalanceListener通知消费者，这允许他们完成必要的应用程序级逻辑，例如状态清除，手动偏移提交等

它也允许消费者通过使用assign(Collection)手动分配指定分区，如果使用手动指定分配分区，那么动态分区分配和协调消费者组将失效。

订阅一组topic，当调用poll(long)时，消费者将自动加入到消费者组中。只要持续调用poll，消费者将一直保持可用，并继续从分配的分区中接收数据。此外，消费者向服务器定时发送心跳。如果消费者崩溃或无法再session.timeout.ms配置的时间内发送心跳，则消费者就被视为死亡，并且其分区将被重新分配。

还有一种可能，消费者可能遇到活锁的情况，它持续的发送心跳，但是没有处理。为了预防消费者在这总情况下一直拥有分区，我们使用max.poll.interval.ms活跃监测机制。在此基础上，如果你调用的poll的频率大于最大间隔，则客户端将主动地离开组，以便其他消费者接管该分区。发生这种情况时，你会看到offset提交失败（ 调用commitSync（）引发的CommitFailedException ）。这是一种安全机制，保障只有活动成员能够提交offset。所以要留在组中，你必须持续调用poll。

消费者提供两种配置设置来控制poll循环：

max.poll.interval.ms: 增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll（long）返回的消息，通常返回的消息都是一批），缺点是此值越大将会延迟组重新平衡。

max.poll.records:此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的

对于消息处理时间不可预测地情况，这些选项是不够的。 处理这种情况的推荐方法是将消息处理移到另一个线程中，让消费者继续调用poll。 但是必须注意确保已提交的offset不超过实际位置。另外，你必须禁用自动提交，并只有在线程完成处理后才为记录手动提交偏移量。 还要注意， 你需要pause暂停分区，不会从poll接收到新消息，让线程处理完之前返回的消息（如果你的处理能力比拉取消息的慢，那创建新线程将导致机器内存溢出）。

实例：

public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers","xxxxxxxxxx:9093"); props.put("group.id","test-6");//消费者组，只要group.id相同，就属于同一个消费者组 props.put("enable.auto.commit","true");//自动提交offset props.put("auto.commit.interval.ms","1000"); // 自动提交时间间隔 props.put("max.poll.records","5"); // 拉取的数据条数 props.put("session.timeout.ms","10000"); // 维持session的时间。超过这个时间没有心跳 就会剔出消费者组 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("auto.offset.reset", "earliest"); KafkaConsumer consumer = new KafkaConsumer<>(props); // 可以写多个topic consumer.subscribe(Arrays.asList("my-topic")); while (true){ ConsumerRecords records = consumer.poll(5000); for (ConsumerRecord record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } System.out.println("处理了一批数据！"); } }

配置说明：

bootstrap.servers: 集群是通过配置bootstrap.servers指定一个或多个broker。不用指定全部的broker，它将自动发现集群中的其余的borker（最好指定多个，万一有服务器故障）

enable.auto.commit: 自动提交偏移量，如果设置了自动提交偏移量,下面这个设置就必须要用到了。

auto.commit.interval.ms:自动提交时间间隔，和自动提交偏移量配合使用

max.poll.records:控制从 broker拉取的消息条数

poll(long time): 当消费者获取不到消息时，就会使用这个参数，为了减轻无效的循环请求消息，消费者会每隔long time的时间请求一次消息，单位是毫秒。

session.timeout.ms: broker通过心跳机器自动检测消费者组中失败的进程，消费者会自动ping集群，告诉进群它还活着。只要消费者能够做到这一点，它就被认为是活着的，并保留分配给它分区的权利，如果它停止心跳的时间超过session.timeout.ms,那么就会认为是故障的，它的分区将被分配到别的进程。

auto.offset.reset:这个属性很重要，一会详细讲解

这里说明一下auto.commit.interval.ms以及何时提交消费者偏移量，经过测试：

设置props.put("auto.commit.interval.ms","60000");

自动提交时间为一分钟，也就是你在这一分钟内拉取任何数量的消息都不会被提交消费的当前偏移量，如果你此时关闭消费者（一分钟内），下次消费还是从和第一次的消费数据一样，即使你在一分钟内消费完所有的消息，只要你在一分钟内关闭程序，导致提交不了offset，就可以一直重复消费数据。

设置props.put("auto.commit.interval.ms","3000");

但是在消费过程中设置sleep。

public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers","xxxxxxxxxxxx:9093"); props.put("group.id","test-6");//消费者组，只要group.id相同，就属于同一个消费者组 props.put("enable.auto.commit","true");//自动提交offset props.put("auto.commit.interval.ms","100000"); // 自动提交时间间隔 props.put("max.poll.records","5"); // 拉取的数据条数 props.put("session.timeout.ms","10000"); // 维持session的时间。超过这个时间没有心跳 就会剔出消费者组 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("auto.offset.reset", "earliest"); // KafkaConsumer consumer = new KafkaConsumer<>(props); // 可以写多个topic consumer.subscribe(Arrays.asList("my-topic")); while (true){ ConsumerRecords records = consumer.poll(5000); for (ConsumerRecord record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } try { Thread.sleep(5000L); System.out.println("等待了5秒了！！！！！！！！！！！！开始等待15秒了"); Thread.sleep(5000L); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("处理了一批数据！"); } }

这里如果你消费了第一批数据，在执行第二次poll的时候，关闭程序也不会提交偏移量，只有在执行第二次poll的时候才会把上一次的最后一个offset提交上去。

auto.offset.reset讲解：

auto.offset.reset的值有三种：earliest，latest，none，代表者不同的意思

earliest： 当各分区下有已经提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费，最常用的值 latest： 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 none： topic各分区都存在已提交的offset时，从offset后开始消费，只要有一个分区不存在已提交的offset，则抛出异常 ！！注意：当使用了latest，并且分区没有已提交的offset时，消费新产生的该分区下的数据，其实是把offset的值直接设置到最后一个消息的位置。例如，有个30条数据的demo的topic，各分区无提交offset，使用了latest，再看offset就会发现已经在30的位置了，所以才只能消费新产生的数据！！！！

不需要定时提交偏移量，可以自己控制offset，当消息已经被我们消费过后，再去手动提交他们的偏移量。这个很适合我们的一些处理逻辑。

手动提交offset的方法有两种：分别是commitSync(同步提交) 和commitAsync(异步提交)。两者的相同点，都会将本次poll的一批数据最高的偏移量提交；不同点是commitSync会失败重试，一直到提交成功（如果有不可恢复的原因导致，也会提交失败），才去拉取新数据。而commitAsync则没有重试机制（提交了就去拉取新数据，不管这次的提交有没有成功），故有可能提交失败。

实例：

public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers","XXXXXC:9093"); props.put("group.id","test-11");//消费者组，只要group.id相同，就属于同一个消费者组 props.put("enable.auto.commit","false");//自动提交offset props.put("auto.commit.interval.ms","1000"); // 自动提交时间间隔 props.put("max.poll.records","20"); // 拉取的数据条数 props.put("session.timeout.ms","10000"); // 维持session的时间。超过这个时间没有心跳 就会剔出消费者组 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("auto.offset.reset", "earliest"); KafkaConsumer consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); int i= 0; while (true){ ConsumerRecords records = consumer.poll(5000); for (ConsumerRecord record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); i++; } if (i == 20){ System.out.println("i_num:"+i); // 同步提交 consumer.commitSync(); // 异步提交 // consumer.commitAsync(); }else { System.out.println("不足二十个，不提交"+i); } i=0; } }

这些都是全部提交偏移量，如果我们想更细致的控制偏移量提交，可以自定义提交偏移量：

public static void main(String[] args) throws InterruptedException { Properties props = new Properties(); props.put("bootstrap.servers","XXXXXXXXXX:9093"); props.put("group.id","test-18");//消费者组，只要group.id相同，就属于同一个消费者组 props.put("enable.auto.commit","false");//自动提交offset props.put("auto.commit.interval.ms","1000000"); // 自动提交时间间隔 props.put("max.poll.records","5"); // 拉取的数据条数 props.put("session.timeout.ms","10000"); // 维持session的时间。超过这个时间没有心跳 就会剔出消费者组 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("auto.offset.reset", "earliest"); KafkaConsumer consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); while (true){ ConsumerRecords records = consumer.poll(5000); for (TopicPartition partition : records.partitions()) { List> partitionRecords = records.records(partition); for (ConsumerRecord record : partitionRecords) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset(); consumer.commitAsync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)), new OffsetCommitCallback() { @Override public void onComplete(Map map, Exception e) { for (Map.Entry entry : map.entrySet()){ System.out.println("提交的分区："+entry.getKey().partition()+",提交的偏移量："+entry.getValue().offset()); } } }); } } }

通过消费者Kafka会通过分区分配分给消费者一个分区，但是我们也可以指定分区消费消息，要使用指定分区，只需要调用assign(Collection)消费指定的分区即可：

public static void main(String[] args) throws InterruptedException { Properties props = new Properties(); props.put("bootstrap.servers","XXXXXXXXX:9093"); props.put("group.id","test-19");//消费者组，只要group.id相同，就属于同一个消费者组 props.put("enable.auto.commit","false");//自动提交offset props.put("auto.commit.interval.ms","1000000"); // 自动提交时间间隔 props.put("max.poll.records","5"); // 拉取的数据条数 props.put("session.timeout.ms","10000"); // 维持session的时间。超过这个时间没有心跳 就会剔出消费者组 props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); props.put("auto.offset.reset", "earliest"); KafkaConsumer consumer = new KafkaConsumer<>(props); // 你可以指定多个不同topic的分区或者相同topic的分区 我这里只指定一个分区 TopicPartition topicPartition = new TopicPartition("my-topic", 0); // 调用指定分区用assign，消费topic使用subscribe consumer.assign(Arrays.asList(topicPartition)); while (true){ ConsumerRecords records = consumer.poll(5000); for (TopicPartition partition : records.partitions()) { List> partitionRecords = records.records(partition); for (ConsumerRecord record : partitionRecords) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset(); consumer.commitAsync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)), new OffsetCommitCallback() { @Override public void onComplete(Map map, Exception e) { for (Map.Entry entry : map.entrySet()){ System.out.println("提交的分区："+entry.getKey().partition()+",提交的偏移量："+entry.getValue().offset()); } } }); } } }

一旦手动分配分区，你可以在循环中调用poll。消费者分区任然需要提交offset，只是现在分区的设置只能通过调用assign 修改，因为手动分配不会进行分组协调，因此消费者故障或者消费者的数量变动都不会引起分区重新平衡。每一个消费者是独立工作的（即使和其他的消费者共享GroupId）。为了避免offset提交冲突，通常你需要确认每一个consumer实例的groupId都是唯一的。

注意：

手动分配分区（assgin）和动态分区分配的订阅topic模式（subcribe）不能混合使用。

EI企业智能 Java Kafka 可信智能计算服务 TICS 智能数据

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。