Kafka源码阅读日记——ProducerInterceptor

Kafka做为当前流行的消息中间件，在消息队列、微服务架构、大数据平台等方面有着普遍的应用。若是将平台比做人体，Kafka即是神经系统，负责传递消息。本系列利用碎片化的时间，阅读Kafka源码深刻了解各个模块的原理和实现，不按期更新。文中全部代码均来自https://github.com/apache/kafka

Kafka Producer简单使用示例

KafkaProducer用于将事件从客户端应用发送至Kafka集群。Producer自己是线程安全的，而且多个线程共享单个实例时也会有性能上的提高。如下示例来自org.apache.kafka.clients.producer.KafkaProducer类：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++)
    producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

producer.close();

props变量定义了Producer的属性和基本配置信息：

• bootstrap.servers：kafka服务器地址，事件会发送至该服务器。在生产环境中一般定义多个服务器并用逗号分隔，以防止单个服务器忽然崩溃。

• acks：当事件发送至Kafka集群时，数据在集群内部会有主从备份，acks定义了什么时候能够断定消息发送成功。

  • acks = 0时，Producer在消息发送后不会等待服务器返回结果，马上返回成功。
  • acks = 1时，消息在主（leader）服务器写入后返回成功，不会等待从（follower）服务器备份完成。
  • acks = all时，消息在主从服务器都写入成功后才告知Producer发送成功。
• retries: 当发送失败时，producer自动重发的次数，并非全部的错误均可以触发自动重发，而且自动重发可能致使消息发送顺序错乱，具体信息将在之后的章节介绍
• key.serializer/value.serializer: 全部发送至kafka的数据都是以byte形式存在的，key/value serializer负责将Java实例转化为字节。

使用上述配置初始化proudcer后，咱们能够构建ProducerRecord，这里使用topic，key，value构建消息并调用producer.send方法发送至kafka集群。在程序结束前务必调用producer.close方法，由于默认状况下producer会在内存中batch多个事件，并一块儿发送以增长性能，close方法会强制发送当前内存中未发送的事件。

发送消息

在上述示例中咱们使用了send接口传入并发送ProducerRecord，在实际实现中该方法使用另外一个send接口并传入了null回调函数。Kafka发送消息是异步的，回调函数能够得到发送结果，若发送成功，回调函数能够获得消息的元数据包括topic，partition，offset等。若失败可得到错误信息。

/**
* Asynchronously send a record to a topic. Equivalent to <code>send(record, null)</code>.
* See {@link #send(ProducerRecord, Callback)} for details.
*/
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
    return send(record, null);
}

/**
* Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.
* <p>
* The send is asynchronous and this method will return immediately once the record has been stored in the buffer of
* records waiting to be sent. This allows sending many records in parallel without blocking to wait for the response after each one.
*/
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

一个回调函数的例子：

producer.send(myRecord,
                   new Callback() {
                       public void onCompletion(RecordMetadata metadata, Exception e) {
                           if(e != null) {
                              e.printStackTrace();
                           } else {
                              System.out.println("The offset of the record we just sent is: " + metadata.offset());
                           }
                       }
                   });

拦截器(ProducerInterceptor)

public interface ProducerInterceptor<K, V> extends Configurable {
  /**
  * 消息发送前调用
  */
  public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
  /**
  * 消息发送后，服务器返回结果（成功或错误）时调用
  */
  public void onAcknowledgement(RecordMetadata metadata, Exception exception);

  /**
  * 拦截器关闭时调用
  */
  public void close();
}

每个Producer均可以设置一个或多个拦截器，拦截器容许客户端拦截或修改要发送的消息，经过Properties进行设置：

Properties props = new Properties();
...
props.put("interceptor.classes", "your.interceptor.class.name");

public class KafkaProducer<K, V> implements Producer<K, V> {
  // ... other class members
  private final ProducerInterceptors<K, V> interceptors;

  // Producer构造函数
  KafkaProducer(ProducerConfig config,
                  Serializer<K> keySerializer,
                  Serializer<V> valueSerializer,
                  Metadata metadata,
                  KafkaClient kafkaClient) {
      // ...其余步骤省略
      // 从config中获取拦截器实例，config从properties中构造
      List<ProducerInterceptor<K, V>> interceptorList = (List) (new ProducerConfig(userProvidedConfigs, false)).getConfiguredInstances(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                    ProducerInterceptor.class);
      this.interceptors = new ProducerInterceptors<>(interceptorList);   
  }           
}

拦截器设置完成后，在send方法中进行调用：

@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

ProducerInterceptors

ProducerInterceptors是一个容器类，封装了多个拦截器，onSend方法被producer的send方法调用。

/**
* A container that holds the list {@link org.apache.kafka.clients.producer.ProducerInterceptor}
* and wraps calls to the chain of custom interceptors.
*/
public class ProducerInterceptors<K, V> implements Closeable {
    private final List<ProducerInterceptor<K, V>> interceptors;

    public ProducerInterceptors(List<ProducerInterceptor<K, V>> interceptors) {
        this.interceptors = interceptors;
    }
    
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        ProducerRecord<K, V> interceptRecord = record;
        // 按顺序执行每个拦截器的onSend方法
        for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
            try {
                interceptRecord = interceptor.onSend(interceptRecord);
            } catch (Exception e) {
                // do not propagate interceptor exception, log and continue calling other interceptors
                // be careful not to throw exception from here
                if (record != null)
                    log.warn("Error executing interceptor onSend callback for topic: {}, partition: {}", record.topic(), record.partition(), e);
                else
                    log.warn("Error executing interceptor onSend callback", e);
            }
        }
        return interceptRecord;
    }
    
    /**
    * 1. 当发送的消息被服务器接受并返回时调用
    * 2. 当发送的消息未到达服务器以前就失败时调用（见下方onSendError方法）
    **/
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
            try {
                interceptor.onAcknowledgement(metadata, exception);
            } catch (Exception e) {
                // do not propagate interceptor exceptions, just log
                log.warn("Error executing interceptor onAcknowledgement callback", e);
            }
        }
    }
    
    /**
    * Producer在发送数据前要构建多种不一样的信息，每一步都有可能抛出异常，本方法由producer在遇到异常时调用，
    * TopicPartition记录了topic和partition信息，由producer构建，但若异常发生在其构建以前，该参数为空，所以从record里提取topic和partition数据构建。
    **/
    public void onSendError(ProducerRecord<K, V> record, TopicPartition interceptTopicPartition, Exception exception) {
        for (ProducerInterceptor<K, V> interceptor : this.interceptors) {
            try {
                if (record == null && interceptTopicPartition == null) {
                    interceptor.onAcknowledgement(null, exception);
                } else {
                    if (interceptTopicPartition == null) {
                        interceptTopicPartition = new TopicPartition(record.topic(),
                                record.partition() == null ? RecordMetadata.UNKNOWN_PARTITION : record.partition());
                    }
                    interceptor.onAcknowledgement(new RecordMetadata(interceptTopicPartition, -1, -1,
                                    RecordBatch.NO_TIMESTAMP, Long.valueOf(-1L), -1, -1), exception);
                }
            } catch (Exception e) {
                // do not propagate interceptor exceptions, just log
                log.warn("Error executing interceptor onAcknowledgement callback", e);
            }
        }
    }
}

须要注意的是，若是拦截器抛出异常，程序不会中止，只会写入一个warn级别的日志。而且拦截器链也不会中止执行，而是继续执行下一个拦截器。若是下一个拦截器依赖于上一个的结果，那么最终获得的数据可能不正确。