kafka：(2) 生产者

一、向Kafka发送消息的主要步骤

• 我们从创建一个 ProducerRecord 对象开始，ProducerRecord对象需要包含目标主题和要发送的内容。我们还可以指定键或分区。在发送 ProducerRecord对象时，生产者要先把键和值对象序列化成字节数组，这样它们才能够在网络上传输 。
• 接下来，数据被传给分区器。如果之前在ProducerRecord对象里指定了分区，那么分区器就不会再做任何事情，直接把指定的分区返回。如果没有指定分区，那么分区器会根据 ProducerRecord对象的键来选择一个分区 。选好分区以后 ，生产者就知道该往哪个主题和分区发送这条记录了。
• 紧接着，这条记录被添加到一个记录批次里，这个批次里的所有消息会被发送到相同的主题和分区上。有一个独立的线程负责把这些记录批次发送到相应的 broker 上。
• 服务器在收到这些消息时会返回一个响应。如果消息成功写入 Kafka，就返回一个 RecordMetaData 对象，它包含了主题和分区信息，以及记录在分区里的偏移量。如果写入失败，则会返回一个错误。生产者在收到错误之后会尝试重新发送消息，几次之后如果还是失败，就返回错误信息。

　　

二、整体架构

　　消息在通过 send()方法发往 broker 的过程中，有可能需要经历拦截器(lnterceptor）、序列化器（Serializer）和分区器（Partitioner）的一系列作用之后才能被真正地发往broker。

• 拦截器：生产者拦截器既可以用来在消息发送前做一些准备工作，比如按照某个规则过滤不符合要求的消息、修改消息的内容等，也可以用来在发送回调逻辑前做一些定制化的需求，比如统计类工作。
• 序列化器：把对象转换成字节数组才能通过网络发送给Kafka。在对侧，消费者需要用反序列化器把从Kafka中收到的字节数组转换成相应的对象。
• 分区器：拦截器 一般不是必需的，而序列化器是必需的。消息经过序列化之后就需要确定它发往的分区，如果消息 ProducerRecord 中指定了 partition 字段，那么就不需要分区器的作用，因为 partition 代表的就是所要发往的分区号。如果消息 ProducerRecord 中没有指定 partition 字段，那么就需要依赖分区器，根据 key 这个字段来计算 partition 的值。分区器的作用就是为消息分配分区。在默认分区器DefaultPartitioner的实现中，如果 key 不为 null，那么默认的分区器会对 key 进行哈希（采用 MurmurHash2 算法 ，具备高运算性能及低碰撞率），最终根据得到的哈希值来计算分区号，拥有相同 key 的消息会被写入同一个分区。如果 key 为 null，那么消息将会以轮询的方式发往主题内的各个可用分区。如果key不为null，那么计算得到的分区号会是所有分区中的任意一个，如果key为null并且有可用分区时，那么计算得到的分区号仅为可用分区中的任意一个，两者是有区别的。

　　

　　那么在此之后又会发生什么呢？

　　整个生产者客户端由两个线程协调运行，这两个线程分别为主线程和Sender线程（发送线程）。在主线程中由KafkaProducer创建消息，然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器（RecordAccumulator，也称为消息收集器）中。Sender 线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。RecordAccumulator主要用来缓存消息以便Sender线程可以批量发送，进而减少网络传输的资源消耗以提升性能。

三、发送消息的3种方式

• 发送并忘记（fire-and-forget）：我们把消息发送给服务器，但并不关心它是否正常到达。大多数情况下，消息会正常到达，因为kafka是高可用的，而且生产者会自动尝试重发。不过这种方式有时候也会丢失一些消息。
• 同步发送：我们使用send()方法发送消息，它会返回一个Future对象。调用get()方法进行等待，就可以知道消息是否发送成功。

ProducerRecord<String,String> record = new ProducerRecord<String, String>("CustomerCountry","West","France");
try{
  RecordMetadata recordMetadata = producer.send(record).get();
}catch(Exception e){
  e.printStackTrace()；
}

　　调用 Future对象的get() 方法等待 Kafka 响应。如果服务器返回错误，get() 方法会抛出异常，如果没有发生错误，我们会得到 RecordMetadata 对象，可以用它获取消息的偏移量。

• 异步发送：我们调用send()方法，并指定一个回调函数，服务器在返回响应时调用该函数。

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("CustomerCountry", "Huston", "America");
producer.send(producerRecord,new DemoProducerCallBack());

class DemoProducerCallBack implements Callback {
  @Override
  public void onCompletion(RecordMetadata metadata, Exception exception) {
    if(exception != null){
      exception.printStackTrace();;
    }
  }
}

　　为了使用回调，需要一个实现了org.apache.kafka.clients.producer.Callback接口的类，这个接口只有一个 onCompletion方法。如果 kafka 返回一个错误，onCompletion 方法会抛出一个非空(non null)异常。

　　生产者是可以使用多线程来发送消息的，刚开始的时候可以使用单个消费者和单个线程。如果需要更高的吞吐量，可以在生产者数量不变的前提下增加线程数量。如果这样做还不够，可以增加生产者数量。

　　KafkaProducer一般会发生两类错误。其中一类是可重试错误，这类错误可以通过重发消息来解决。比如对于连接错误，可以通过再次建立连接来解决，"无主"错误则可以通过重新为分区选举首领来解决。KafkaProducer 可以被配置成自动重试，如果在多次重试后仍无法解决问题，应用程序会收到一个重试异常。另一类错误无法通过重试来解决，比如消息过大异常。对于这类错误，KafkaProducer 不会进行任何重试，直接抛出异常。

四、生产者的配置

• acks：指定了必须要有多少个分区副本收到消息，生产者才会认为消息写入是成功的。

　　如果 acks=0 生产者在成功写入消息之前不会等待任何来自服务器的响应。如果当中出现了问题，导致服务器没有收到消息，生产者就无从得知，消息也就丢失了。可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。
　　如果 acks=1 只要集群的首领节点收到消息，生产者就会收到一个来自服务器的成功响应。如果消息无法到达首领节点（比如首领节点崩溃，新的首领还没选举出来）生产者收到一个错误响应，为了避免数据丢失，生产者会重发消息。
　　如果 ack=all 只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。这种模式是最安全的，它可以保证不止一个服务器收到消息，就算有服务器发生崩溃，整个集群仍然可以运行。不过，它的延迟比 acks=1 时更高，因为我们要等待不止一个服务器节点接收消息。

• buffer.memory：用来设置生产者内存缓冲区的大小，用来缓冲发送到服务器的消息。
• retries：生产者从服务器收到的错误有可能是临时性的错误（例如分区找不到首领），retries 参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待 100ms，可以通过 retry.backoff.ms 参数来改变这个时间间隔。一般情况下，因为生产者会自动进行重试，所以就没必要在代码逻辑里处理那些可重试的错误，只需要处理那些不可重试的错误或重试次数超出上限的情况。
• batch.size：当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，当批次被填满，批次里的所有消息会被发送出去。
• linger.ms：指定生产者在发送批次之前等待更多消息加入批次的时间。kafkaProducer 会在批次填满或 linger.ms 达到上限时把批次发送出去。
• max.in.flight.requests.per.connection：指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提高吞吐量。把它设为1可以保证消息是按照发送的顺序写入服务器的，即使发送了重试。
• timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms：request.timeout.ms 指定了生产者发送数据时等待服务器返回响应的时间，metadata.fetch.timeout.ms 指定了生产者在获取元数据时等待服务器返回响应的时间。如果等待超时，那么生产者要么重试发送，要么返回一个错误。timeout.ms 指定了 broker 等待同步副本返回消息确认的时间，与 acks 的配置相匹配，如果在指定时间内没有收到同步副本的确认，那么broker 就会返回一个错误。
• max.request.size：用于控制生产者发送的请求大小。它可以指能发送的单个消息的最大值，也可以指单个请求里所有消息总的大小。例如，假设这个值为 1MB，或者生产者可以在单个请求里发一个批次，该批次有100个消息，每个消息大小为10KB，另外，broker 对可接收的消息最大值也有自己的限制（message.max.bytes）所有两边的配置最好可以匹配，避免生产者发送的消息被 broker 拒接。
• receive.buffer.bytes 和 send.buffer.bytes：分别指 TCP socket 接收和发送数据包的缓冲区大小。

五、顺序保证

　　Kafka可以保证同一个分区里的消息是有序的。即如果生产者按照一定的顺序发送消息，broker就会按照这个顺序把它们写入分区，消费者也会按照同样的顺序去读取它们。
　　如果某些场景要求消息是有序的，那么消息是否写入成功也是很关键的，所以不建议把retires设为0 。可以把max.in.flight.request.per.connection设为1，这样在生产者尝试发送第一批消息时，就不会有其他的消息发送给broker。不过这样会严重影响生产者的吞吐量，所以只有在对消息的顺序有严格要求的情况下才能这么做。

六、分区

　　ProducerRecord 对象包含了目标主题、键和值。Kafka 的消息是一个个键值对，ProducerRecord对象可以只包含目标主题和值，键可以设置为默认的null，不过大多数的应用程序会用到键。键有两个用途：可以作为消息的附加信息，也可以用来决定消息该被写到主题的哪个分区。拥有相同键的消息将被写到同一个分区。如果一个进程只从一个主题的分区读取数据，那么具有相同键的所有记录都会被该进程读取。
　　如果键为null，并且使用了默认的分区器，那么记录将被随机地发送到主题内各个可用的分区上。分区器使用轮询（Round Robin）算法将消息均衡地分布到各个分区上。
　　如果键不为空，并且使用了默认分区器，那么 Kafka 会对键进行散列（使用Kafka 自己的散列算法）然后根据散列值把消息映射到特定的分区上。这里的关键之处在于，同一个键总是被映射到同一个分区上，所以在进行映射时，我们会使用主题所有的分区，而不仅仅是可用的分区。意味着，如果写入数据的分区时不可用的，那么就会发生错误。
　　只有在不改变主题分区数量的情况下，键与分区之间的映射才能保持不变。一旦主题增加了新的分区，这些就无法保证了，旧数据仍然留在旧分区，但新记录可能被写到其他分区上。如果要使用键来映射分区，最好在创建主题的时候就把分区规划好，而且永远不要增加新分区。

　　自定义分区： 

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.record.InvalidRecordException;
import org.apache.kafka.common.utils.Utils;
public class BananaPartitioner implements Partitioner {
　　public void configure(Map<String, ?> configs) {} 
    
　　public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
    int numPartitions = partitions.size();
    if ((keyBytes == null) || (!(key instanceOf String))) ➋
    throw new InvalidRecordException("We expect all messages to have customer name as key")

    if (((String) key).equals("Banana"))
    return numPartitions; // Banana总是被分配到最后一个分区
    
　　 // 其他记录被散列到其他分区
    return (Math.abs(Utils.murmur2(keyBytes)) % (numPartitions - 1))
    }    

　　public void close() {}
}

• Partitioner接口包含了configure、partition和close这3个方法，不应该直接在partition方法里硬编码客户的名字，应该通过configure方法传进来。
• 这里只接受字符串作为键，如果不是字符串，就抛出异常。