掌握这些玩转Kafka客户端-Producer_org.apache.kafka.clients.producer.*;

 客户端版本gradle引用：compile "org.apache.kafka:kafka-clients:0.10.2.1"

org.apache.kafka.clients：producer和consumer相关类
org.apache.kafka.common：公共模块
org.apache.kafka.server.policy ：只有一个类，创建一个topic强制指定的策略类          .

生产者发送消息逻辑

org.apache.kafka:kafka-clients：0.10.21==>>org.apache.kafka.clients.producer.KafkaProducer.send()//359行

1.拦截器过滤消息
org.apache.kafka.clients.producer.internals.ProducerInterceptor.onSend()//57行

2.first make sure the metadata for the topic is available，同步集群元数据

3.序列化 key和value，序列化方法和编码都可以配置

4.获取分区
org.apache.kafka:kafka-clients：0.10.21==>>org.apache.kafka.clients.producer.KafkaProducer.partition()//470行
    指定分区，直接返回
    未指定分区，指定分区序列化key：Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions
    未指定分区，未指定分区序列化key
        轮询分区数字，然后通过可达分区(集群可达分区数，可变)取余数：Utils.toPositive(nextValue) % availablePartitions.size()
        没有可达分区：Utils.toPositive(nextValue) % numPartitions

5.alidate that the record size is not too large

6.将记录放到消息记录收集器里
org.apache.kafka.clients.producer.internals.RecordAccumulator.append()//160行 

7.如果改批次已经满了，新建一个批次然后将记录追加到批次里；如果一个批次没有，新建一个批次然后将记录追加到批次里
if (result.batchIsFull || result.newBatchCreated) {

    //唤醒发送线程
    this.sender.wakeup();
 }

Producer 发送消息流程详解

1. 构造kafkaProducer对象
   1. 很多默认配置，只配置这两个就可以运行了
      1. bootstrap.servers：连接broker集群的地址，配置一个或多个
      2. key.serializer、value.serializer：配置类的全类路径字符
   2. 很多配置的key容易配错，可以用ProducerConfig类里的所有配置常量来配置，很好的方法
   3. producer实例是线程安全的
2. 拦截器处理
   1. 拦截器可以对发送的所有消息做自定义的改动
   2. 比如对消息做敏感字段过滤，可以实现但是不建议在这儿做；比如做消息的统计
   3. 可以通过实现ProducerInterceptor来实现自己的定制逻辑,主要实现下面三个方法
      1. onSend(略)：消息序列化和计算分区之前回调用此方法
      2. onAcknowledgment(略)：消息ack应答之前 or 消息发送失败调用此方法，此方法先于自定义的Callback执行，建议逻辑简单点，因为在IO线程中执行这个方法会消耗性能
      3. close()：在关闭拦截器时执行清理工作
      4. 三个方法抛出的异常都上传给上层，只是做日志记录
3. 序列化器对key和value进行序列化
   1. 序列化的目的就是为了网络传输的时候高效
   2. 默认包里提供了很多序列化类：ByteArray、ByteBuffer、Bytes、Double、Integer、Long、String
   3. 可以自定义自己的序列化类，只要实现Serializer接口就可以了，使用的时候也就是创建一个Producer的时候制定序列化类使用你自定义的
   4. 还可以通过第三方的序列化工具完成你的高级需求：JSON、Thrift、ProtoBuffer、Protostuff等
4. 分区器选出一个目标分区
   1. 如果你指定了分区，用消息的key通过一个hash来计算分区，一个高性能碰撞低的算法MurmurHash2实现的
   2. 如果没指定分区，通过集群的元数据选出所有可用的，也就是活着的分区
      1. 获取轮询分区数字，然后通过可达分区(集群可达分区数，可变)取余数：Utils.toPositive(nextValue) % availablePartitions.size()
      2. 没有可达分区，直接通过numPartitions取余：Utils.toPositive(nextValue) % numPartitions
   3. 注意：Hash获取的分区有可能已经死亡了，轮询的方式获取的时候是从存活的分区中选一个，二者有区别
5. 将消息放入消息收集器里
   1. 将消息做应用层的批量缓冲
   2. 消息收集器里，一个分区对应一个双端队列Deque
   3. Deque里的每个元素叫ProducerBatch，可以理解成一个批次，一个批次可以容纳多一个消息
   4. 怎么创建一个producerBatch还有更复杂的流程
6. 异步线程Sender将把消息收集器里producerBatch转换成Request
   1. 异步线程Sender从消息收集器RecordAccumulator中取出需要发送的ProducerBatch
   2. 将ProducerBatch转换成Rquest对象，ProducerBatch是应用层的缓冲对象，Rquest是网络层的缓冲对象
7. 将Requet对象交给NetworkClient
   1. 将所有的ProducerBatch按照broker节点分类，因为有的ProducerBatch需要发送到Broker1有的需要发送到Broker2
   2. 将转换好的Rquest按照Broker节点再此放到InFlightRequets队列中，这里网络层再此排队缓冲
8. 将Requet对象交给Selector准备发送
   1. Requet最终要发送到某个分区的Leader上，也就是要和Leader所在的Broker节点通信，这里如果一个Requet过去Broker就开启一个线程处理的化非常耗费资源，所以Kafka自己实现了一套NIO协议，以此提高网络通信的性能
9. Selector将Request发送给对应的Broker节点
   1. Selector将选择目前可使用的channel做数据交互
10. Broker对象Response返回给NetwokrClient
    1. Response返回Broker已经成功接受消息后，InFlightRequests中一个请求才算成功发送了
11. NetworkClient清理缓冲

 

对KafkaProducer的思考

1. 为了提高Producer和Broker的网络IO效率
   1. Kafka自己实现了NIO网络模块
   2. Producer端做了小微量的批量提交缓冲
      1. 这个思想在应用开发中也可以参考，比如我们调用小米做AppPush的时候就此处的思想做了批量提交来减少IO次数，效果很明显
   3. 网络通信，消息的大小也是很重要的，Kafka支持7种自带序列化工具、可以自定义、可用第三方的
2. 做缓冲最重要就是考虑什么时候刷新
   1. 达到空间多大了刷新，空间
   2. 达到间隔多久了刷新，时间
3. KafkaProducer端做了两次缓冲
   1. 应用层的消息收集器RecordAccumulator，需要考虑的配置
      1. RecordAccumulator整个缓冲空间大小，buffer.memory 默认是32M
      2. Producer.send()方法的阻塞时间，max.block.ms 默认是60s
      3. Producerbatch的大小，batch.size 默认16KB，但是也不一定必须是16KB，大于它也可以
         1. BufferPool的设计可以对Producerbatch的空间做复用，如果Producerbatch的空间不是16KB就不能复用，用完立即回收
         2. 当消息的大小大于16KB，就创建一个大于16K不的producerBatch来容纳消息此时的Producerbatch空间不能复用，这要注意次细节
      4. 一个ProducerBatch什么时候出队放入InFIightRequests队列
         1. 当ProducerBatch的空间满了，肯定要出队
         2. 生产者发送到ProducerBatch之前等待更多的消息加入ProducerBatch，用linger.ms控制，默认0：每次不用等，有消息直接放入ProducerBatch
   2. 网络层请求层InFIightRequests队列缓冲，需要考虑的配置
      1. 限制每个连接缓冲的最大请求数：max.in.flight.requests.per.connection 默认是5
      2. 每个Requst的最大大小，max.request.size 默认是1M，所以一个Request可以容纳多个ProducerBatch
      3. 网络通信是Socket通信还有两个参数配置Socket接受和发送的缓冲取大小
         1. receive.buffer.bytes:默认32KB，配置-1表示使用操作系统默认配置
         2. send.buffer.bytes:默认128KB，配置-1表示使用操作系统默认配置
      4. Requet请求的超时时间，request.timeout.ms:默认30s，超时了可以重试，注意这个时间需要比服务端的replica.lag.time.max.ms要大，这样可以减少重试引起的重复概率
   3. 还可以使用压缩配置参数：compression.type:支持gzip、snappy、lz4
   4. 客户端网络连接的最大空间时间：connetions.max.idle.ms：默认9分钟
   5. 元数据更新的默认间隔时间 5分钟:metadata.max.age.ms
   6. 说的最多的一个参数ack最后来说
      1. 设置成1：表示producer发完就走
      2. 设置成0：表示producer必须等到分区的leader节点收到消息才能走
      3. 设置成-1或all：表示producer必须等到所有正在同步的副本（ISR）都收到了消息才能走
   7. 怎么保证消息有序
      1. 需要有序的消息放到同一个分区
      2. InFIightRequests队列的缓冲个数设置成0，如果设置成2，第一个request1失败了重试放到request2的后面消息就乱序了，
      3. 默认kafka的producer端重试为0，所以只要保证消息生产和消费在一个分区就有序了
      4. 如果重试次数不为0，就必须InFIightRequests的缓冲个输
   8. 怎么保证绝对不丢消息
      1. ack设置成-1或all并不能保证消息绝对不丢
         1. 如果ISR中只有一个存活了，也就是leader，当leader收到消息后，然后挂掉，然后再选一个Leader的时候可能消息就丢了
         2. 所以服务端的ISR个数最好配置成大于1才能和ACK一起保证消息不丢失