卡夫卡］Kafka源代码分析之生产者流程解读-生产者整体流程

对于Producer来说，其实就是几部分。

• 初始化、发送流程、缓冲区

初始化流程

设置分区器

// 设置分区器
  this.partitioner = config.getConfiguredInstance(
                    ProducerConfig.PARTITIONER_CLASS_CONFIG,
                    Partitioner.class,
                    Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));

设置重试时间，默认100ms，如果配置Kafka可以重试，retries制定重试次数，retryBackoffMs指定重试的间隔

 long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);

获取Key和Value的序列化器

      // 序列化器
      if (keySerializer == null) {
          this.keySerializer = config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                                                                                   Serializer.class);
          this.keySerializer.configure(config.originals(Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId)), true);
      } else {
          config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
          this.keySerializer = keySerializer;
      }
      
      if (valueSerializer == null) {
          this.valueSerializer = config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                                                                                     Serializer.class);
          this.valueSerializer.configure(config.originals(Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId)), false);
      } else {
          config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
          this.valueSerializer = valueSerializer;
      }

拦截器

    // 设置拦截器
     List<ProducerInterceptor<K, V>> interceptorList = (List) config.getConfiguredInstances(
             ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
             ProducerInterceptor.class,
             Collections.singletonMap(ProducerConfig.CLIENT_ID_CONFIG, clientId));
     if (interceptors != null)
         this.interceptors = interceptors;
     else
         this.interceptors = new ProducerInterceptors<>(interceptorList);

其他参数

	   // 设置最大消息为多大,默认是1M 默认是16384
       this.maxRequestSize = config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG);
       // 设置缓存大小 默认是32M 默认是33554432 RecordAccumulator=32MB
       this.totalMemorySize = config.getLong(ProducerConfig.BUFFER_MEMORY_CONFIG);
       // 设置压缩类型 可以提升性能
       this.compressionType = CompressionType.forName(config.getString(ProducerConfig.COMPRESSION_TYPE_CONFIG));
        this.accumulator = new RecordAccumulator(logContext,

       // 因为是通过缓冲区发送消息的,所以需要消息累计器
       RecordAccumulator.PartitionerConfig partitionerConfig = new RecordAccumulator.PartitionerConfig(
           enableAdaptivePartitioning,
           config.getLong(ProducerConfig.PARTITIONER_AVAILABILITY_TIMEOUT_MS_CONFIG)
       );

初始化元数据

 // 初始化集群元数据
   if (metadata != null) {
       this.metadata = metadata;
   } else {
       this.metadata = new ProducerMetadata(retryBackoffMs,
               config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),
               config.getLong(ProducerConfig.METADATA_MAX_IDLE_CONFIG),
               logContext,
               clusterResourceListeners,
               Time.SYSTEM);
       this.metadata.bootstrap(addresses);
   }

创建Sender线程，其中包含一个重要的网络组件NetWorkClient

	// 创建sender线程
   this.sender = newSender(logContext, kafkaClient, this.metadata);
    // 线程name
    String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
    // 封装起来 设置为守护线程 并启动
    this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
    // 线程启动
    this.ioThread.start();

发送消息流程

发送消息的过程

    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // 执行拦截器逻辑
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        return doSend(interceptedRecord, callback);
    }

先执行拦截器，可以发现就是遍历拦截器，然后执行对应的onSend()方法。当我们想增加一个拦截器，直接实现对应的接口，重写onSend()方法，然后Kafka就会调用我们的onSend方法。通过提供一个拓展点进行使用。

    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        ProducerRecord<K, V> interceptRecord = record;
        for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
            try {
                interceptRecord = interceptor.onSend(interceptRecord);
            } catch (Exception e) {
            }
        }
        return interceptRecord;
    }

从Kafka Broker集群获取元数据metadata

// 从broker获取元数据
	clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), nowMs, maxBlockTimeMs);

对key和value进行序列化，调用对应的serialize的方法。

	  byte[] serializedKey;
      try {
          // 选择对应的序列化进行操作
          serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
      } catch (ClassCastException cce) {
      }
      
      byte[] serializedValue;
      try {
          serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());
      } catch (ClassCastException cce) {
         
      }

 	// 选择具体的分区
	int partition = partition(record, serializedKey, serializedValue, cluster);
 // 消息缓存到RecoredAccumulator
	result = accumulator.append(record.topic(), partition, timestamp, serializedKey,
                    serializedValue, headers, appendCallbacks, remainingWaitMs, false, nowMs, cluster);

     // 消息发送的条件
     // 缓冲区数据大小达到batch.size 或者linnger.ms达到上限后 唤醒sneder线程。
     if (result.batchIsFull || result.newBatchCreated) {
         log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), appendCallbacks.getPartition());
         this.sender.wakeup();
     }

Sender线程

	runOnce();

    long pollTimeout = sendProducerData(currentTimeMs);

缓冲区、

这篇讲解很详细 https://www.cnblogs.com/rwxwsblog/p/14754810.html

生产者核心参数配置

bootstrap.servers：连接Broker配置，一般就是xxxx:9092

key.serializer 和 value.serializer：对key和value进行序列化器，可以自定义，一般就是String方式

buffer.memory：RecordAccumulator 缓冲区总大小，默认32m。

batch.size： 消息会以batch的方式进行发送，这是一批数据的大小 默认是16K

linger.ms：发送消息的时机，如果没有达到batch.size or linger.ms的时间就会发送 默认是0ms 立即发送

acks： 0: 不落盘 1:只有leader落盘 -1(all) : leader和所有从节点持久化成功 默认是-1

max.in.flight.requests.per.connection：允许最多没有返回 ack 的次数，默认为 5

retries: 消息发送失败时，系统重发消息 默认值 2147483647

retry.backoff.ms：两次重试间隔 默认是100ms

enable.idempotence: 开启幂等性 默认true

compression.type: 压缩格式 默认是none