生产者向 Kafka 发送消息的执行流程

（1）生产者要往 Kafka 发送消息时，需要创建 ProducerRecoder,代码如下：

ProducerRecord<String,String> record 
      = new ProducerRecoder<>("CostomerCountry","Precision Products","France");
      try{
      producer.send(record);
      }catch(Exception e){
        e.printStackTrace();
      }

（2）ProducerRecoder 对象会包含目标 topic，分区内容，以及指定的 key 和 value,

在发送 ProducerRecoder 时，生产者会先把键和值对象序列化成字节数组

，然后在网络上传输。

（3）生产者在将

消息

发送到某个 Topic ，需要经过拦截器、序列化器和分区器（Partitioner）。

（4）如果消息 ProducerRecord 没有指定 partition 字段，那么就需要依赖分区器，根据 key 这个字段来计算 partition 的值。分区器的作用就是为消息分配分区。

1. 若没有指定分区，且消息的 key 不为空，则使用 murmur 的 Hash 算法（非加密型 Hash 函数，具备高运算性能及低碰撞率）来计算分区分配。
2. 若没有指定分区，且消息的 key 也是空，则用轮询的方式选择一个分区。

（5）分区选择好之后，会将消息添加到一个记录批次中，这个批次的所有消息都会被发送到相同的 Topic 和 partition 上。然后会有一个独立的线程负责把这些记录批次发送到相应的 broker 中。

（6）broker 接收到 Msg 后，会作出一个响应。如果成功写入 Kafka 中，就返回一个 RecordMetaData 对象，它包含

Topic

和

Partition

信息，以及记录在分区的

offset

。

（7）若写入失败，就返回一个错误异常，生产者在收到错误之后尝试重新发送消息，几次之后如果还失败，就返回错误信息。