Kafka - 2 客户端工作机制1

public static final String GROUP_ID_CONFIG = "group.id";
private static final String GROUP_ID_DOC = "A unique string that identifies the consumer group this consumer belongs to. This property is required if the consumer uses either the group management functionality by using <code>subscribe(topic)</code> or the Kafka-based offset management strategy.";

public static final String INTERCEPTOR_CLASSES_CONFIG = "interceptor.classes";
public static final String INTERCEPTOR_CLASSES_DOC = "A list of classes to use as interceptors. Implementing the <code>org.apache.kafka.clients.producer.ProducerInterceptor</code> interface allows you to intercept (and possibly mutate) the records received by the producer before they are published to the Kafka cluster. By default, there are no interceptors.";

public class MyInterceptor implements ProducerInterceptor {
  //发送消息时触发
  @Override
  public ProducerRecord onSend(ProducerRecord producerRecord) {
      System.out.println("prudocerRecord : " + producerRecord.toString());
      return producerRecord;
  }

  //收到服务端响应时触发
  @Override
  public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
      System.out.println("acknowledgement recordMetadata:"+recordMetadata.toString());
  }

  //连接关闭时触发
  @Override
  public void close() {
      System.out.println("producer closed");
  }

  //整理配置项
  @Override
  public void configure(Map<String, ?> map) {
      System.out.println("=====config start======");
      for (Map.Entry<String, ?> entry : map.entrySet()) {
          System.out.println("entry.key:"+entry.getKey()+" === entry.value: "+entry.getValue());
      }
      System.out.println("=====config end======");
  }
}

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.roy.kfk.basic.MyInterceptor");

public static final String KEY_SERIALIZER_CLASS_CONFIG = "key.serializer";
public static final String KEY_SERIALIZER_CLASS_DOC = "Serializer class for key that implements the <code>org.apache.kafka.common.serialization.Serializer</code> interface.";
public static final String VALUE_SERIALIZER_CLASS_CONFIG = "value.serializer";
public static final String VALUE_SERIALIZER_CLASS_DOC = "Serializer class for value that implements the <code>org.apache.kafka.common.serialization.Serializer</code> interface.";

public static final String KEY_DESERIALIZER_CLASS_CONFIG = "key.deserializer";
public static final String KEY_DESERIALIZER_CLASS_DOC = "Deserializer class for key that implements the <code>org.apache.kafka.common.serialization.Deserializer</code> interface.";
public static final String VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer";
public static final String VALUE_DESERIALIZER_CLASS_DOC = "Deserializer class for value that implements the <code>org.apache.kafka.common.serialization.Deserializer</code> interface.";

public static final String PARTITIONER_CLASS_CONFIG = "partitioner.class";
private static final String PARTITIONER_CLASS_DOC = "Determines which partition to send a record to when records are produced. Available options are: " + 
"<ul>" + 
"<li>If not set, the default partitioning logic is used. This strategy send records to a partition until at least batch.size bytes is produced to the partition." + 
"It works with the strategy:"+
"<ol>"+
"<li>If no partition is specified but a key is present, choose a partition based on a hash of the key.</li>"+
"<li>If no partition or key is present, choose the sticky partition that changes when at least batch.size bytes are produced to the partition.</li></ol></li>"+
"<li><code>org.apache.kafka.clients.producer.RoundRobinPartitioner</code>:" +
"A partitioning strategy where each record in a series of consecutive records is sent to a different partition, regardless of whether the 'key' is provided or not, until partitions run out and the process starts over again."+
"Note: There's a known issue that will cause uneven distribution when a new batch is created. See KAFKA-9965 for more detail."+
"</li></ul>"+
"<p>Implementing the <code>org.apache.kafka.clients.producer.Partitioner</code> interface allows you to plug in a custom partitioner.";

public interface Partitioner extends Configurable, Closeable {
  int partition(String var1, Object var2, byte[] var3, Object var4, byte[] var5, Cluster var6);

  void close();

  /** @deprecated */
  @Deprecated
  default void onNewBatch(String topic, Cluster cluster, int prevPartition) {
  }
}

/** @deprecated */
@Deprecated
public class DefaultPartitioner implements Partitioner {
  private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();

  public void configure(Map<String, ?> configs) {
  }

  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
      return this.partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());
  }

  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster, int numPartitions) {
      return keyBytes == null ? this.stickyPartitionCache.partition(topic, cluster) : BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);
  }

  public void close() {
  }

  public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
      this.stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
  }
}

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
  // 【情况 1：指定分区】
  // 如果你在发消息时代码里写死了分区：new ProducerRecord(topic, partition, key, value)
  // 优先级最高，直接用，不走任何计算逻辑。
  if (record.partition() != null) {
      return record.partition();
  } 
  
  // 【情况 2：外包模式 (Plugin Path)】
  // 这里就是重点！如果你在配置里设置了 partitioner.class = ...
  // 不管你是设置了自定义的类，还是显式设置了那个过时的 DefaultPartitioner，
  // this.partitioner 都不会是 null。
  else if (this.partitioner != null) {
      // 于是这里会走“接口调用”，进入你指定的类里去算。
      int customPartition = this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
      if (customPartition < 0) {
          throw new IllegalArgumentException(String.format("The partitioner generated an invalid partition number: %d. Partition number should always be non-negative.", customPartition));
      } else {
          return customPartition;
      }
  } 
  
  // 【情况 3：直营模式 (Internal Fast Path)】
  // 这就是现在默认的高效路径！
  // 如果你没配 partitioner.class，this.partitioner 就是 null，直接进这里。
  else {
    // 逻辑：
    // 1. 如果有 Key (serializedKey != null) 且没忽略 Key：直接调用内置静态方法算 Hash。
    // 2. 如果没有 Key：返回 -1。
    return serializedKey != null && !this.partitionerIgnoreKeys ? 
            BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size()) : 
            -1;
  }
}

public class RoundRobinPartitioner implements Partitioner {
  // 定义了一个自增器
  // key: Topic value: 当前的partition (原子类实现并发)
  private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap();

  public void configure(Map<String, ?> configs) {
  }

  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    // 得到当前的分区原始索引值
    int nextValue = this.nextValue(topic);
    // 获取当前Topic的总分区数
    List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
    if (!availablePartitions.isEmpty()) {
      // 优先在“活着”的分区里轮询
      // 取模运算，转到对应分区
      int part = Utils.toPositive(nextValue) % availablePartitions.size();
      return ((PartitionInfo)availablePartitions.get(part)).partition();
    } else {
      // 如果所有分区都挂了（或者不可用），那就死马当活马医，在“所有”分区里轮询
      int numPartitions = cluster.partitionsForTopic(topic).size();
      return Utils.toPositive(nextValue) % numPartitions;
    } 
  }

  private int nextValue(String topic) {
    // 如果是第一次见这个 topic，创建一个新的计数器（从0开始）
    AtomicInteger counter = (AtomicInteger)this.topicCounterMap.computeIfAbsent(topic, (k) -> new AtomicInteger(0));
    // 返回当前值，并加 1
    return counter.getAndIncrement();
  }

  public void close() {
  }
}

public static final String PARTITION_ASSIGNMENT_STRATEGY_CONFIG = "partition.assignment.strategy";
private static final String PARTITION_ASSIGNMENT_STRATEGY_DOC = "A list of class names or class types, ordered by preference, of supported partition assignment strategies that the client will use to distribute partition ownership amongst consumer instances when group management is used." + 
"Available options are:"+
"<ul>"+
"<li>"+
"<code>org.apache.kafka.clients.consumer.RangeAssignor</code>:"+
"Assigns partitions on a per-topic basis."+
"</li>"+
"<li>"+
"<code>org.apache.kafka.clients.consumer.RoundRobinAssignor</code>:"+
"Assigns partitions to consumers in a round-robin fashion.</li>"+
"<li>"+
"<code>org.apache.kafka.clients.consumer.StickyAssignor</code>:"+
"Guarantees an assignment that is maximally balanced while preserving as many existing partition assignments as possible.</li>"+
"<li>"+
"<code>org.apache.kafka.clients.consumer.CooperativeStickyAssignor</code>: Follows the same StickyAssignor logic, but allows for cooperative rebalancing.</li></ul>"+
"<p>The default assignor is [RangeAssignor, CooperativeStickyAssignor], which will use the RangeAssignor by default, but allows upgrading to the CooperativeStickyAssignor with just a single rolling bounce that removes the RangeAssignor from the list.</p>"+
"<p>Implementing the <code>org.apache.kafka.clients.consumer.ConsumerPartitionAssignor</code> interface allows you to plug in a custom assignment strategy.</p>";