当我们学习Kafka时，通常写kafka producer api如下面简单的代码：
public static void main(String[] args) {
//配置信息
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "localhost:9092");
//设置数据key和value的序列化处理类
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
//创建生产者实例
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
//发送记录
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if(Objects.isNull(exception))
System.out.println("success");
}
});
producer.close();
}
从上面的API可以看到其中一个非常重要的对象：
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
这个就是生产者初始化对象，这个对象在始化时做了大量的操作，到底做了什么，我们必须从源码才可以看到，有句话说的好，源码面前无秘密，想深入掌握 一个开源的系统，底层还要了解的。
由于目前我们用的kafka版本是1.0.1,我们就用这个版本来简单聊一聊，其中大的版本这块也大基本修改很小
首先，我们找KafkaProducer这个类,最终查看它的构造方法如下：
private KafkaProducer(
ProducerConfig config,
Serializer<K> keySerializer,
Serializer<V> valueSerializer) {
Map<String, Object> userProvidedConfigs = config.originals();
this.producerConfig = config;
this.time = Time.SYSTEM;
/**
* KafkaProducer初始化的时候会涉及到哪些内部的核心组件，默认情况下，一个jvm内部，
* 如果你要是搞多个KafkaProducer的话，
* 每个都默认会生成一个client.id，producer-自增长的数字，producer-1
*/
String clientId = config.getString(ProducerConfig.CLIENT_ID_CONFIG);
if (clientId.length() <= 0)
clientId = "producer-" + PRODUCER_CLIENT_ID_SEQUENCE.getAndIncrement();
this.clientId = clientId;
/**
* 获取transactionalId
* 这块是查看用户端是否配置了事务id,关于这个事务我们后面详细的说明
*/
String transactionalId = userProvidedConfigs.containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG) ?
(String) userProvidedConfigs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG) : null;
LogContext logContext;
if (transactionalId == null)
logContext = new LogContext(String.format("[Producer clientId=%s] ", clientId));
else
logContext = new LogContext(String.format("[Producer clientId=%s, transactionalId=%s] ", clientId, transactionalId));
log = logContext.logger(KafkaProducer.class);
log.trace("Starting the Kafka producer");
...
}
上面其实就是查看用户端是否设置了client_id,如果没有设置系统会每个都默认会生成一个client.id，producer-自增长的数字，producer-1，producer-2等
- 接着我们看下面的代码逻辑
/**
* 核心组件：Partitioner，
* 后面用来决定，你发送的每条消息是路由到Topic的哪个分区里去的
*/
this.partitioner = config.getConfiguredInstance(ProducerConfig.PARTITIONER_CLASS_CONFIG, Partitioner.class);
//retry.backoff.ms 发送消息失败，重试间隔时间
long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG);
我们知道Kafka发送每条消息都会有一个路由操作的，其实就是被分配到哪个分区里去，这个操作就是分区器Partitioner,用户端可以自定义这个分区器，如下：
//ProducerConfig.PARTITIONER_CLASS_CONFIG对应的值：partitioner.class
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,new MyPartitioner);
如果我们没有配置这个，那么系统会分配给我们一个默认的分区器：DefaultPartitioner，这个分区器后面我们在讲发送消息时候会详细讲一下，这里看到在初始化的时候，把这个解析出来。
在看上面的源码 retry.backoff.ms 这个参数，下面会用到在说
- 接下来我们看下面的逻辑，这不就是序列化器嘛，这其实就是初始化了我们常说的key,value，keySerializer ，valueSerializer ，正如上面我们API里设置的
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
下面主要是获取我们设置的参数值
/**
* 核心组件：序列化器key value
*/
if (keySerializer == null) {
this.keySerializer = ensureExtended(config.getConfiguredInstance(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
Serializer.class));
this.keySerializer.configure(config.originals(), true);
} else {
config.ignore(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
this.keySerializer = ensureExtended(keySerializer);
}
if (valueSerializer == null) {
this.valueSerializer = ensureExtended(config.getConfiguredInstance(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
Serializer.class));
this.valueSerializer.configure(config.originals(), false);
} else {
config.ignore(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
this.valueSerializer = ensureExtended(valueSerializer);
}
接下是我们看常常说的第三个组件，拦截器
/**
* 核心组件：序列化组件，拦截器组件
*/
this.interceptors = interceptorList.isEmpty() ? null : new ProducerInterceptors<>(interceptorList);
ClusterResourceListeners clusterResourceListeners = configureClusterResourceListeners(keySerializer, valueSerializer, interceptorList, reporters);
到这里，我们常说的三个组件，都初始化完成
分区器Partitioner
序列化器Serializer
拦截器Interceptor
由于本部分内容较多，我们接下来分节讲解，下一节讲解非常的几个核心组件：
Metadata
RecordAccumulator
NetworkClient
网友评论