kafka依赖_SpringBoot系列：关于Kafka消息发送的事儿

世界上最快的捷径&＃xff0c;就是脚踏实地&＃xff0c;本文已收录【架构技术专栏】关注这个喜欢分享的地方。

开源项目&＃xff1a;

• 分布式监控(Gitee GVP最有价值开源项目 )&＃xff1a;https://gitee.com/sanjiankethree/cubic
• 个人主站&＃xff1a;http://www.jiagoujishu.com

0、简述

Spring Boot 版本&＃xff1a;2.3.4.RELEASE

随着大数据的发展&＃xff0c;目前Kafka可以说在我们项目中的使用是越来越多了。其高性能的特点也是满足了我们大部分的场景&＃xff0c;所以对于学习Kafka的兼容使用也是一件很重要的事情。

下面我们从几个点来说&＃xff1a;

• 发送消息
• 发送回调
• 实现原理
• 异步和同步

1、添加依赖

 org.springframework.kafkaspring-kafka


2、添加配置

在Spring Boot 中kafka的配置属性都是spring.kafka.* 开头的,最简配置如下(application.properties中 )

spring.kafka.bootstrap-servers&＃61;localhost:9092
spring.kafka.consumer.group-id&＃61;myGroup


3、发送代码

Spring Boot kafka 依然沿用老套路 XXXTemplate&＃xff0c;所以这里发送自然就使用了KafkaTemplate

&＃64;Component
public class Producer {

   private static final Logger logger &＃61; LoggerFactory.getLogger(Producer.class);

    &＃64;Resource
    private KafkaTemplate kafkaTemplate;public void send(String msg) {
         kafkaTemplate.send("testTOPIC", msg);
    }
}

4、参数说明

ListenableFuture> sendDefault(V data);
ListenableFuture> sendDefault(K key, V data);
ListenableFuture> sendDefault(Integer partition, K key, V data);
ListenableFuture> sendDefault(Integer partition, Long timestamp, K key, V data);
ListenableFuture> send(String topic, V data);
ListenableFuture> send(String topic, K key, V data);
ListenableFuture> send(String topic, Integer partition, K key, V data);
ListenableFuture> send(String topic, Integer partition, Long timestamp, K key, V data);
ListenableFuture> send(ProducerRecord record);
ListenableFuture> send(Message> message);

使用KafkaTemplate.send 会有很多不同的发送参数&＃xff0c;这里说明下:

• topic : 填写要发送的topic名称
• partition &＃xff1a; 要发送的分区id,从0开始
• timestamp&＃xff1a;时间戳
• key&＃xff1a;消息的key
• data&＃xff1a;消息数据
• ProducerRecord&＃xff1a;消息的封装类&＃xff0c;包含了上面的参数
• Message> &＃xff1a;Spring自带的Message封装类&＃xff0c;包含消息和消息头

5、发送回调

Spring Boot KafkaAutoConfiguration 为我们提供了处理消息回调的handler&＃xff0c;以供我们来处理结果。成功调用onSuccess&＃xff0c;失败调用onError&＃xff0c;增加如下类&＃xff1a;

&＃64;Component
public class KafkaSendResultHandler implements ProducerListener {

    private static final Logger log &＃61; LoggerFactory.getLogger(KafkaSendResultHandler.class);

    &＃64;Override
    public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
        log.info("Message send success : " &＃43; producerRecord.toString());
    }

    &＃64;Override
    public void onError(ProducerRecord producerRecord, Exception exception) {
        log.info("Message send error : " &＃43; producerRecord.toString());
    }
}


6、实现原理

从上面来看&＃xff0c;我们基本两三行代码就完成了kafka消息的发送&＃xff0c;那他们到底是怎么加载实现的呢。

熟悉Spring Boot 的小伙伴想必也能猜到&＃xff0c;基于其扩展SPI的机制&＃xff0c;spring-boot-autoconfigure包下一定会有一个KafkaAutoConfiguration配置类。

1、KafkaAutoConfiguration

&＃64;Configuration(proxyBeanMethods &＃61; false)
&＃64;ConditionalOnClass(KafkaTemplate.class)
&＃64;EnableConfigurationProperties(KafkaProperties.class)
&＃64;Import({ KafkaAnnotationDrivenConfiguration.class, KafkaStreamsAnnotationDrivenConfiguration.class })public class KafkaAutoConfiguration {
 
  
}


通过KafkaAutoConfiguration我们可以看出几件事情

1、&＃64;ConditionalOnClass(KafkaTemplate.class) 表示我们必须依赖了spring-kafka 包才会加载KafkaAutoConfiguration

2、此类配置了KafkaProperties属性供我们使用

3、这里在加载时帮我们引入了两个类KafkaAnnotationDrivenConfiguration 提供消费注解支持 和KafkaStreamsAnnotationDrivenConfiguration 提供stream 注解支持

2、KafkaTemplate

这里就不大段贴代码了&＃xff0c;可以去看完整的KafkaAutoConfiguration。

  &＃64;Bean
 &＃64;ConditionalOnMissingBean(KafkaTemplate.class)public KafkaTemplate, ?> kafkaTemplate(ProducerFactory kafkaProducerFactory,ProducerListener kafkaProducerListener,ObjectProvider messageConverter) {
  KafkaTemplate kafkaTemplate &＃61; new KafkaTemplate<>(kafkaProducerFactory);
  messageConverter.ifUnique(kafkaTemplate::setMessageConverter);
  kafkaTemplate.setProducerListener(kafkaProducerListener);
  kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());return kafkaTemplate;
 }&＃64;Bean&＃64;ConditionalOnMissingBean(ProducerListener.class)public ProducerListener kafkaProducerListener() {return new LoggingProducerListener<>();
 }&＃64;Bean&＃64;ConditionalOnMissingBean(ProducerFactory.class)public ProducerFactory, ?> kafkaProducerFactory() {return factory;
 }

这里摘取了KafkaTemplate 主要相关的方法&＃xff0c;关键的几个点如下&＃xff1a;

• 三个方法使用了&＃64;ConditionalOnMissingBean 注解&＃xff0c;根本原因就是为了方便我们进行扩展而存在的
• kafkaProducerListener 方法是为了在调用doSend 方法是构建Callback 使用的&＃xff0c;方便我们来监控发送成功或失败的信息(KafkaTemplate 的305行buildCallback)
• ProducerFactory 是真正用于创建producer的&＃xff0c;如果配置了 transactionIdPrefix 那就代表开启了producer 对于事物的支持。如果开启了事物那就会先从本地ThreadLocal 中获取producer&＃xff0c;拿不到才去创建。(KafkaTemplate 的 341 行 getTheProducer)
• 这里如果配置了spring.kafka.producer.transaction-id-prefix还会创建一个KafkaTransactionManager事务管理器

加载流程&＃xff1a;

1、因为我们加入spring-kafka jar&＃xff0c;所以在启动的时候会通过SPI 机制加载到 KafkaAutoConfiguration

2、这时配置类通过&＃64;ConditionalOnMissingBean 发现我们没有独立配置 KafkaTemplate时&＃xff0c;会依次加载默认的ProducerListener和ProducerFactory来构建KafkaTemplate

发送流程&＃xff1a;

1、调用KafkaTemplate.send(String topic, &＃64;Nullable V data)方法

2、调用KafkaTemplate 内部 doSend(ProducerRecord producerRecord)方法

3、调用KafkaTemplate 内部 getTheProducer() 方法&＃xff0c;如果是事物发送就从Threadlocal 获取&＃xff0c;否则创建一个Producer

4、构造SettableListenableFuture回调

5、调用最终的发送方法KafkaProducer 内的 doSend(ProducerRecord record, Callback callback)

7、异步和同步发送

通过KafkaTemplate 的源码我们可以发现&＃xff0c;其实发送消息都是采用异步发送的。

KafkaTemplate会把我们传入的参数封装成ProducerRecord&＃xff0c;然后调用doSend方法&＃xff0c;源码如下&＃xff1a;

public ListenableFuture> send(String topic, K key, &＃64;Nullable V data) {
        ProducerRecord producerRecord &＃61; new ProducerRecord(topic, key, data);return this.doSend(producerRecord);
    }

而doSend(producerRecord)方法先检查了下是否开启事物&＃xff0c;调用this.getTheProducer()获取到 producer。

后面主要是构造了一个SettableListenableFuture 回调&＃xff0c;最后在使用KafkaProducer.send(ProducerRecord record, Callback callback) 进行数据发送&＃xff0c;返回一个Future

protected ListenableFuture> doSend(ProducerRecord producerRecord) {if (this.transactional) {
            Assert.state(this.inTransaction(), "No transaction is in process; possible solutions: run the template operation within the scope of a template.executeInTransaction() operation, start a transaction with &＃64;Transactional before invoking the template method, run in a transaction started by a listener container when consuming a record");
        }
        Producer producer &＃61; this.getTheProducer();this.logger.trace(() -> {return "Sending: " &＃43; producerRecord;
        });
        SettableListenableFuture> future &＃61; new SettableListenableFuture();
        producer.send(producerRecord, this.buildCallback(producerRecord, producer, future));if (this.autoFlush) {this.flush();
        }this.logger.trace(() -> {return "Sent: " &＃43; producerRecord;
        });return future;
    }

同步发送消息

因为在某些业务场景下我需要同步发送消息&＃xff0c;实现其实也很简单。因为返回了一个Future 所以我们只需要调用get方法就行了

public void syncSend() throws ExecutionException, InterruptedException {
        kafkaTemplate.send("demo", "test sync message").get();
    }


往期推荐

Spring Boot 系列&＃xff1a;日志动态配置详解

Spring Boot 系列&＃xff1a;最新版优雅停机详解

(最新 9000字)  Spring Boot 配置特性解析

Spring Boot 知识清单(一)SpringApplication

何时用多线程&＃xff1f;多线程需要加锁吗&＃xff1f;线程数多少最合理&＃xff1f;