在Kafka历史变迁中,一共有两大版本的生产者客户端,一个是0.9.x之前的scala客户端;另一个是之后引入的Java客户端。当这并不代表Kafka不具有多语言的支持性。实际上,常用的语言如C/C++,Python,Go等语言都有Kafka的客户端,只不过不由官方进行维护
本文主要聊一下Java客户端的生产者模块
在说明Kafka生产者相关的内容之前,先上一个生产者Demo:
public class KafkaProducerAnalysis { publ static final String brokerList = "localhost:9092"; public static final String topic = "topic-demo"; public static Properties int Config () { Properties props= new Properties() ; props.put ("bootstrap.servers", brokerList) ; props.put ("key.serializer","org apache kafka.common.serialization.StringSerializer") ; props.put("value ser alizer ","org.apache.kafka.common.serialization.StringSerializer"); props.put ("client .id", " producer.client.id.demo") ; return props; } public static void main (String [] args) { Properties props = inittConfig(); KafkaProducer<String , String> producer= new KafkaProducer<>(props); ProducerRecord<String, String> record =new ProducerRecord<> (topic,”hello , Kafka 1” ) ; try { producer.send(record); }catch(Exception e){ e.printStackTrace(); } } } 复制代码
一个最简单的Producer就是这样子了
Kafka的生产者中,消息发送其实分为三种: 发后既忘( fire-and-forg 〕、同步( sync )及异步 (async) 。
一开始的Demo用的就是发后既忘的模式,调用send之后,Producer直接返回,并不关心消息是否正确到达。在大多数情况下,这样的方式都没有问题,但是某些特殊情况下(比如出现了某些不可重试的异常)就会造成消息丢失,而Producer是无法感知的。所以发送既忘的方式,性能最好,吞吐量最高,但可靠性也是最差的。
同步的方法其实也很简单,这里就要先说明一下send方法的方法签名:
public Future<RecordMetadata> send(ProducerRecord<K, V> record) public Future<RecordMetadata> send(ProducerRecord<K, V> record,Callback callback) 复制代码
可以看到,该方法的返回值并不是void,而是一个Future接口实现类对象,有过多线程编程的童鞋应该已经知道它的作用了,就是可以以阻塞的方式获取异步执行的结果,这里就不赘述了。
然后再来看看RecordMetadata,该对象内部包含了对应消息的一些元数据,比如消息的主题,分区号,在分区中偏移量等信息,如果需要这个发送结果的话,可以调用Future的get()来获取该信息,如:
Future<RecordMetadata> future = producer.send(record); RecordMetadata rm = future.get() //阻塞方法 //do something 复制代码
如果在发送过程中出现错误而发送出错,并且重试次数已达到上限(可通过retries参数来设置),get方法就是抛出异常,所以可以try-catch这段代码,在catch模块中进行重试等操作
同步发送吞吐量最低,但可靠性最高
异步发送的操作也很简单,直接调用send的一个重载方法:
producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata , Excepti on exception){ if (exception ! = null) { exception.printStackTrace(); } else { System.out.println(metadata . topic() +”-” + metadata . partition() + ”:” ÷metadata . offset()) ; } }); 复制代码
消息在从Producer客户端发送到broker的过程中,有可能会经过拦截器,序列化器,分区器等组件后才会到达broker,下面主要介绍一下拦截器和分区器。序列化器的话,其实就是对对象进行序列化和反序列化而已
先上一个生产者客户端整体架构图:
拦截器是Kafka在0.10.0.0版本引入的一个功能。拦截器其实分为生产者拦截器和消费者拦截器,这里主要讲生产者拦截器。
在Kafka中,生产者拦截器的定义接口为:org.apache.kafka.clients.producer.Producerlnterceptor
其中的方法有三个:
public ProducerRecord<K, V> onSend (ProducerRecord<K, V> record ); public void onAcknowledgement(RecordMetadata metadata , Exception exception); public void close() ; 复制代码
onSend方法就是消息进行序列化和计算分区之前执行的一个逻辑,可以看到直接把record对象发过来了,所以理论上可以对消息实体进行任意改变(当然,topic,key等信息还是不要改比较好,有可能会影响其他部分)
onAcknowledgement方法在消息应答或者消息发送失败时被调用,先于异步发送中的Callback执行,并且该方法是由Kafka Producer的IO线程执行的,所以该方法不应该执行太久,或者影响吞吐量
生产者拦截器Demo:
public class ProducerinterceptorPrefix implements Producerinterceptor<String, String>{ @Override public ProducerRecord<String,String> onSend(ProducerRecord<String,String> record){ String newStr = "prefix-"+record.value(); return new ProducerRecord<>(record.topic,record.partition(),record.timestamp(),record.key, newStr,record.headers()); } //这个方法可以用来统计消息发送成功率之类的 @Override public void onAcknowledgement(RecordMetadata recordMetadata,Exception e){} @Override public void close(){} @Override public void configure(Map<String,?>map){} } 复制代码
怎么生效呢?好办:
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG ,ProducerinterceptorPrefix.class.getName()); 复制代码
这样,Producer发送消息:msg 而消费者接受到的就是prefix-msg了
经历了拦截器,然后就是序列化器,下一步,如果Producer在发送信息时没有显式指定分区的号,接下来就会进入到是分区器了,在这一步,主要是进行分区的计算
Kafka中的分区器实现接口是:org.apache kafka.clients. producer.Partitioner
其下有两个方法:
public int partition(String topic,Object key,byte[] keyBytes,Object value,byte[] valueBytes,Cluster cluster) ; public void close() ; 复制代码
partition方法就是返回分区的下标。Kafka提供的默认分区器是org.apache.kafka.clients.producer.internals.DefaultPartitioner,在partition方法中采用了MurmurHash2算法
自定义分区器Demo:
public class MyPartitioner implements Partitioner { @Override public int partition(String topic,Object key,byte[] keyBytes,Object value,byte[] valueBytes,Cluster cluster) { List<Partitionlnfo> partitions= cluster .partitionsForTopic(topic) ; int numPartitions =partitions.size(); if (null == keyBytes) { return counter.getAndincrement () numPartitions }else{ return Utils.toPositive(Utils.murmur2(keyBytes)%numPartitions; } } @Override publ raid close () { } @Override public raid configure (Map<String , ?> configs) {} 复制代码
怎么能生效呢?同样的,在Properties里面配置:
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,MyPartitioner. class.getName() ) ; 复制代码
这是个啥玩意?Kafka Producer为了减少网络通信次数,提高发送效率,并不会对每一个信息都进行一次单独的发送,而是整合到一个缓冲区里面,进行统一的发送。类比于TCP发送端的Nagle算法
RecordAccumulator缓冲区的大小由参数buffer.memory决定,默认为33554432B,即32MB。如果发送速度过快导致缓冲区被占满,send方法将阻塞,最长时间由max.block.ms设置,默认值为60000,即60s。超时则抛出异常。
RecordAccumulator内以分区为单位,为每个分区维护一个双端队列,消息进来的时候,从尾端插入队列,Sender线程从队列头取出消息并发送。
队列中的结点为:ProducerBatch 即队列签名为:Deque .哎?ProducerBatch是啥,为什么不是ProducerRecord,它们的关系是啥?实际上,ProducerBatch包含多个ProducerRecord,消息发送的基本单位是ProducerBatch。
当一个ProducerRecord到达RecordAccumulator时,会先找到自己分区对应的队列,然后查看队列尾端的ProducerBatch是否能容纳这个ProducerRecord,如果可以,直接写入;否则,创建一个新的ProducerBatch再写入