【README】

1，本文主要关注 KafkaTemplate的重点方法，并非全部方法；

2，KafkaTemplate 底层依赖于 DefaultKafkaProducerFactory ，关于 DefaultKafkaProducerFactory 的介绍，refer2

spring-kafka整合:DefaultKafkaProducerFactory默认kafka生产者工厂介绍_PacosonSWJTU的博客-CSDN博客【1】类描述类描述：单例共享 Producer 实例的 ProducerFactory 实现。此实现将为每次 createProducer() 调用时提供的 Map 配置和可选的 Serializer 实现返回相同的 Producer 实例（如果未启用事务）。如果您使用的序列化器没有参数构造函数并且不需要设置，那么最简单的方法是在传递给 DefaultKafkaProducerFactory 构造函数的配置中针对 ProducerConfig.KEY_SERIALIZER_CLASS_Chttps://blog.csdn.net/PacosonSWJTU/article/details/121306370

【1】KafkaTemplate 类说明

用于执行高级操作的模板。当与 DefaultKafkaProducerFactory 一起使用时，模板是线程安全的。生产者工厂和 org.apache.kafka.clients.producer.KafkaProducer 确保这一点；

public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationContextAware, BeanNameAware,ApplicationListener<ContextStoppedEvent>, DisposableBean {

【1.1】构造方法

使用提供的生产者工厂和 autoFlush 设置创建一个实例。

如果您已将生产者的 linger.ms 配置为非默认值并希望立即在此模板上发送操作，无论该设置如何，又或者您希望阻塞直到服务器根据acs属性确认已收到消息，需要把autoFlush设置为true。

如果 configOverrides 不为 null 或不为空，则将使用合并的生产者属性创建一个新的 DefaultKafkaProducerFactory，这些属性在提供的工厂属性之后进行覆盖。

public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,@Nullable Map<String, Object> configOverrides) {Assert.notNull(producerFactory, "'producerFactory' cannot be null");this.autoFlush = autoFlush;this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;// 是否自定义生产者工厂 this.customProducerFactory = configOverrides != null && configOverrides.size() > 0;if (this.customProducerFactory) {Map<String, Object> configs = new HashMap<>(producerFactory.getConfigurationProperties());
// 覆盖工厂属性 configs.putAll(configOverrides); // 创建新的 DefaultKafkaProducerFactoryDefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(configs, producerFactory.getKeySerializerSupplier(), producerFactory.getValueSerializerSupplier());
// 设置物理关闭生产者的超时时间  newFactory.setPhysicalCloseTimeout((int) producerFactory.getPhysicalCloseTimeout().getSeconds());
// 设置是否分区 newFactory.setProducerPerConsumerPartition(producerFactory.isProducerPerConsumerPartition());
// 设置是否 每个线程创建一个 生产者； newFactory.setProducerPerThread(producerFactory.isProducerPerThread());
// 新工厂赋值this.producerFactory = newFactory;} else {this.producerFactory = producerFactory;}
// 是否开启kafka事务 this.transactional = this.producerFactory.transactionCapable();
}

【1.2】发送消息方法（非常重要）

发送消息有很多方法，大致分为两类；

send();
doSend()；

【1.2.1】send() 发送消息

有4个外观方法，使用的都是默认topic；

@Override
public ListenableFuture<SendResult<K, V>> sendDefault(@Nullable V data) {return send(this.defaultTopic, data);
}@Override
public ListenableFuture<SendResult<K, V>> sendDefault(K key, @Nullable V data) {return send(this.defaultTopic, key, data);
}@Override
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, @Nullable V data) {return send(this.defaultTopic, partition, key, data);
}@Override
public ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, @Nullable V data) {return send(this.defaultTopic, partition, timestamp, key, data);
}
@Override
public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);return doSend(producerRecord);
}

可以看到，最后还是调用了底层的 doSend() 方法；

【1.2.2】doSend() 方法

5个 doSend() 方法的外观方法，这5个方法对 topic ，分区，消息key，时间戳，消息value 进行了重载

@Override
public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);return doSend(producerRecord);
}@Override
public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);return doSend(producerRecord);
}@Override
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);return doSend(producerRecord);
}@Override
public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key,@Nullable V data) {ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, timestamp, key, data);return doSend(producerRecord);
}@Override
public ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record) {Assert.notNull(record, "'record' cannot be null");return doSend(record);
}

底层 doSend() 定义如下：

protected ListenableFuture<SendResult<K, V>>

doSend(final ProducerRecord<K, V> producerRecord)

protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {// 获取生产者 final Producer<K, V> producer = getTheProducer(producerRecord.topic());this.logger.trace(() -> "Sending: " + producerRecord);final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();Object sample = null;if (this.micrometerEnabled && this.micrometerHolder == null) {this.micrometerHolder = obtainMicrometerHolder();}if (this.micrometerHolder != null) {sample = this.micrometerHolder.start();}// 发送消息 Future<RecordMetadata> sendFuture =producer.send(producerRecord, buildCallback(producerRecord, producer, future, sample));// May be an immediate failure （注意，这里可能马上失败，或有运行时异常抛出）if (sendFuture.isDone()) { try {sendFuture.get(); // 这里调用get会阻塞，如果发送没有完成的话 }catch (InterruptedException e) {Thread.currentThread().interrupt();throw new KafkaException("Interrupted", e);}catch (ExecutionException e) {throw new KafkaException("Send failed", e.getCause()); // NOSONAR, stack trace}}if (this.autoFlush) { // 自动刷新 flush();}this.logger.trace(() -> "Sent: " + producerRecord);return future;
}

【代码解说】

step1，调用了 getTheProducer() 获取生产者；

关于 DefaultKafkaProducerFactory.createProducer() 可以参见以下博文，因篇幅，本文不再赘述；

spring-kafka整合:DefaultKafkaProducerFactory默认kafka生产者工厂介绍_PacosonSWJTU的博客-CSDN博客

step2，producer.send(producerRecord, buildCallback(producerRecord, producer, future, sample)) ；调用了 buildCallback(...) 构建回调对象；

非事务模式，则关闭生产者；

由 DefaultKafkaProducerFactory 可知，生产者是 CloseSafeProducer，其包裹了原生 kafka生产者；所以调用了 CloseSafeProducer.close() 方法；

step3，自动刷新缓存 flush()；

如果 ProducerFactory 提供单例生产者（例如 DefaultKafkaProducerFactory），则调用此方法才有意义。

public void flush() {Producer<K, V> producer = getTheProducer();try {producer.flush();}finally {closeProducer(producer, inTransaction());}
}protected void closeProducer(Producer<K, V> producer, boolean inTx) {if (!inTx) { // 非事务才关闭 producer.close(this.closeTimeout);}
}

【2】KafkaTemplate 发送消息与生产者复用

我们再次 follow了 DefaultKafkaProducerFactory的 doCreateProducer() 方法；

第1次因为发送消息新建了 producer；

第2次再发送消息时，因为producer 不为null；所以直接取走；

同时 synchronized同步块可以避免并发问题；

发送消息后，是否关闭生产者，可以参考【小结】

【小结】

通过分析 KafkaTemplate.doSend() 消息发送分发，我们可以看到，

每发送一条消息，如果抛出异常的话，则会关闭kafka生产者，否则不会关闭生产者；原因参见

spring-kafka整合:DefaultKafkaProducerFactory默认kafka生产者工厂介绍_PacosonSWJTU的博客-CSDN博客https://blog.csdn.net/PacosonSWJTU/article/details/121306370中的章节【4.5.1】；

spring-kafka整合:KafkaTemplate-kafka模板类介绍相关推荐

spring boot整合Quartz 在Job类中注入其他对象报空指针异常java.lang.NullPointerException at com.sxt.quartz.QuartzDemo.e
情况在Job 类中注入RedisTemplate 报空指针异常原因:我们使用JobDetailFactoryBean 帮我们创建对象实际上调用的是AdaptableJobFactory 下的这个方 ...
druid监控页面_Spring boot学习（四）Spring boot整合Druid
前言在上一篇博客中我们介绍了Spring boot配置Mybatis,但是并没有配置连接池,这在实际开发过程中肯定是不切实际的,多次的数据库连接会给程序和数据库都带来没必要的负担,这一篇博客我将介绍 ...
spring boot 2.x 系列 —— spring boot 整合 kafka
文章目录一.kafka的相关概念: 1.主题和分区 2.分区复制 3. 生产者 4. 消费者 5.broker和集群二.项目说明 1.1 项目结构说明 1.2 主要依赖二. 整合 kafka 2 ...
Spring Boot 整合——Spring Boot整合kafka整合
Spring Boot 整合之前的内容项目名称描述地址 base-data-mybatis 整合mybatis-plus(实际上官方教程已经很多,只做了自定义插件) 未完成 base-jpa J ...
kafka专题：kafka单机和集群安装详情，Spring Boot如何整合Kafka
文章目录 1. kafka单机安装 1.1 server.properties核心配置 2. kafka集群安装 2.1 kafka集群可视化界面 3. springboot如何整合kafka 4. ...
最新Kafka教程(包含kafka部署与基本操作、java连接kafka、spring连接kafka以及使用springboot)
最新Kafka教程(包含kafka部署与基本操作.java连接kafka.spring连接kafka以及使用springboot) 欢迎转载,转载请注明网址:https://blog.csdn.net ...
freemarker ftl模板_Spring Boot2 系列教程(十)Spring Boot 整合 Freemarker
今天来聊聊 Spring Boot 整合 Freemarker. Freemarker 简介这是一个相当老牌的开源的免费的模版引擎.通过 Freemarker 模版,我们可以将数据渲染成 HTML ...
java 模板引擎_极简 Spring Boot 整合 Thymeleaf 页面模板
点击"牧码小子"关注,和众多大牛一起成长! 关注后,后台回复 java ,领取松哥为你精心准备的技术干货! 虽然现在慢慢在流行前后端分离开发,但是据松哥所了解到的,还是有一些公司在 ...
Spring Boot整合Thymeleaf模板引擎
转载自 Spring Boot整合Thymeleaf模板引擎什么是Thymeleaf Thymeleaf是一款用于渲染XML.XHTML.HTML5内容的模板引擎.类似Velocity,FreeMa ...

spring-kafka整合:KafkaTemplate-kafka模板类介绍