目录

Spring Boot 整合 Kafka 快速入门

生产者发送消息添加回调监听

生产者发送消息自定义分区

生产者发送消息添加事务管理

消费者批量消费处理消息

消费者消费消息并转发消息

消费者组（Consumer Group）测试

Spring.kafka 配置属性介绍

KafkaTemplate API 介绍

本文环境：Spring Boot 2.1.3.RELEASE + Java JDK 1.8 + Spring Kafka 2.2.4 + Kafka Clients 2.0.1。在 IP 为 192.168.116.128、192.168.116.129 的两台 CentOS 虚拟机上进行了 zookeeper 集群与 kafka 集群，并启动成功。

Spring Boot 整合 Kafka 快速入门

1、Spring 通过 Spring-Kafka 项目的对 Apache Kafka 提供自动配置支持。

2、强烈推荐从以下官网进行学习：

一：导入依赖：pom.xml

<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

内部依赖了：kafka-clients、spring-context、spring-messaging、spring-tx、spring-retry。

二：kafka 配置：application.yml

1、Spring boot 对 kafka 的配置以 spring.kafka.* 开头，官网提供了大量的配置选项，通常情况下，绝大部分选项直接使用默认值即可。

2、比如应用只用来发送消息时，则 spring.kafka.consumer.* 不用配置，应用只用了接收消息时，则 spring.kafka.producer.* 不用配置。比如：

3、本文对比较常用的选项进行配置说明，实际生产中，大部分都不用配置，默认即可。在线配置源码：https://gitee.com/wangmx1993/apache-kafka/blob/master/src/main/resources/application.yml

三：kafka 生产者·发送消息

1、如同使用 JdbcTemplate、RedisTemplate、ElasticsearchTemplate 一样，{@link KafkaAutoConfiguration} 中会自动根据 {@link KafkaProperties} 配置属性读取配置， 然后创建 {@link KafkaTemplate} 实例并添加到 Spring 容器中，在需要使用的时候直接获取使用即可。Spring boot 发送消息官方文档。

2、在线演示源码:java/com/wmx/apachekafka/controller/KafkaProducer.java, 参考其中 sendMessage、sendMsgDefault 方法。

这里我使用一个可视化工具来查看 kafka 服务器上的消息：Kafka 可视化工具 Kafka Tool，能清晰的看到消息所属的主题、以及所在的分区，消息的内容、长度等等。

3、注意事项：

[root@wangmaoxiong001 config]# /home/kafka_2.13-2.6.0/bin/kafka-topics.sh --create --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 --replication-factor 2 --partitions 2 --topic helloWorld
Created topic helloWorld.
[root@wangmaoxiong001 config]# /home/kafka_2.13-2.6.0/bin/kafka-topics.sh --describe --bootstrap-server 192.168.116.128:9092,192.168.116.129:9092 -topic helloWorld
Topic: helloWorld    PartitionCount: 2    ReplicationFactor: 2    Configs: segment.bytes=1073741824
    Topic: helloWorld    Partition: 0    Leader: 129    Replicas: 129,128    Isr: 129,128
    Topic: helloWorld    Partition: 1    Leader: 128    Replicas: 128,129    Isr: 128,129
[root@wangmaoxiong001 config]#

四：kafka 消息者·接收消息

1、在 Spring 组件（Component）中，新建一个方法，然后在方法上标注 @KafkaListener(topics = “someTopic”) 注解即可接收指定主题的消息。Spring boot 接收消息官方文档。

2、注意：在消费者第一次启动表示自己需要监听某些主题后，凡是推送到此主题的消息，都会被监听然后接收，即使生产者推送的时候，消费者已经宕机了，或者消费者服务关闭了，都不会受影响，消费者启动后都会接收未接收的消息。

3、在线演示源码：/src/main/java/com/wmx/apachekafka/controller/KafkaConsumer.java

更多 @KafkaListener 的配置使用可以参考官网：@KafkaListener Annotation

♛♛♛棒棒哒♛♛♛

生产者发送消息添加回调监听

1、生成者发送信息，我们非常关心消息是否发送成功，比如网络断开，或者 kafka 服务器宕机等等，消息发送之后，如果达到发送失败时的重试次数仍然未成功，则发送失败，抛出异常，此时需要有相应的处理，比如记录日志，保存到数据库，后期再次发送等。

2、实现方式是在 send(发送)后添加异步回调监听方法，用于监控消息发送成功或者失败，发送成功则可以记录日志，发送失败则应该有相应的措施，比如延期再发送等。

2、有如下两种实现方式：

3、在线演示源码：https://gitee.com/src/main/java/com/wmx/apachekafka/controller/KafkaProducer.java，参考其中的 sendMessageCallback、sendMessageCallback2 方法。

生产者发送消息自定义分区

1、kafka 中的 topic 可以指定多个分区，多个副本，各个分区中的数据是不同的，一条消息只会进入一个分区，生产者将消息发送到 topic 时，遵循以下策略：

2、实际开发中通常使用上面默认的策略就可以了，因为更关心的是消息的发送与接收，至于消息在 kafka 中存放在哪个分区，其实意义不是很大。

kafkaTemplate.send(String topic, Integer partition, K key, @Nullable V data)

3、如果自己硬是想要自定义分区策略，即自定义消息路由到分区的规则，则可以实现 Partitioner 接口，重写方法，其中 partition 方法的返回值就表示将消息发送到几号分区。在线演示源码

4、然后在application.propertise 中配置自定义分区器：

生产者发送消息添加事务管理

1、很多时候发送消息是需要和系统业务相关联的，比如发送消息后还需要做其它的业务操作，比如增删改查，或者远程调用其它微服务等，当后续操作发生异常时，前面发送给 kafka 的消息是让它发送成功还是回滚取消发送呢？这通常取决于具体情况。

2、使用事务时，必须修改如下两项配置：

方式一：使用 springframework 的 @Transactional注解，此时事务处理完全和平时一样，发生异常，则回滚，消息也不会发送成功：

/**
     * 向指定主题(topic)发送消息：http://localhost:8080/kafka/sendMsgTransactional1?topic=car-infos
     * 与 springframework 框架的事务整合到一起，此时异常处理完全和平时一样，默认所有运行时异常都将回滚。
     *
     * @param topic   ：主题名称，不存在时自动创建，默认1个分区，无副本。主题名称也可以通过配置文件配置，这里直接通过参数传入。
     * @param message ：待发送的消息，如：{"version":1,"text":"后日凌晨三点执行任务"}
     * @return
     */
    @PostMapping("kafka/sendMsgTransactional1")
    @Transactional
    public Map<String, Object> sendMessageTransactional1(@RequestParam String topic, @RequestBody Map<String, Object> message) {
        try {
            logger.info("向指定主题发送信息，带事务管理，topic={},message={}", topic, message);
            String msg = new ObjectMapper().writeValueAsString(message);
            kafkaTemplate.send(topic, msg);
            if ("110".equals(message.get("version").toString())) {
                System.out.println(1 / 0);
            }
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return message;
    }

方式二：使用 KafkaTemplate#executeInTransaction，此时消息发送操作是在本地事务中进行，与全局事务是分离的，外部的异常并不影响内部的发送。

/**
     * 生成者发送消息事务管理方式2：使用 executeInTransaction(OperationsCallback<K, V, T> callback)
     * executeInTransaction：表示执行本地事务，不参与全局事务(如果存在)，即方法内部和外部是分离的，只要内部不
     * 发生异常，消息就会发送，与外部无关，即使外部有 有 @Transactional 注解也不影响消息发送，此时外围有没有 @Transactional 都一样。
     *
     * @param topic
     * @param message
     * @return
     */
    @PostMapping("kafka/sendMsgTransactional2")
    @Transactional(rollbackFor = Exception.class)
    public Map<String, Object> sendMessageTransactional2(@RequestParam String topic, @RequestBody Map<String, Object> message) {
        try {
            logger.info("向指定主题发送信息，带事务管理：topic={},message={}", topic, message);
            String msg = new ObjectMapper().writeValueAsString(message);
            /**
             * executeInTransaction 表示这些操作在本地事务中调用，不参与全局事务（如果存在）
             * 所以回调方法内部发生异常时，消息不会发生出去，但是方法外部发生异常不会回滚，即便是外围方法加了 @Transactional 也没用。
             */
            kafkaTemplate.executeInTransaction(operations -> {
                operations.send(topic, msg);
                if ("120".equals(message.get("version").toString())) {
                    System.out.println(1 / 0);
                }
                return null;
            });
            //如果在这里发生异常，则只要 executeInTransaction 里面不发生异常，它仍旧会发生消息成功
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return message;
    }

本文在线源码：https://gitee.com/wangmx1993/apache-kafka/blob/master/src/main/java/com/wmx/apachekafka/controller/KafkaProducer.java

消费者批量消费处理消息

1、默认一次只消费一条消息，注意所谓的批量消费并不是一定要满足设置的条数后才进行消费，而是最多一次消费那么多消息。修改配置如下：

# 监听器类型，可选值有：SINGLE（单条消费，默认）、BATCH（批量消息）
spring.kafka.listener.type=batch
# 一次调用poll（）时返回的最大记录数，即批量消费每次最多消费多少条消息，注意是最多，并不是必须满足数量后才消费.
spring.kafka.consumer.max-poll-records=10

2、配置的批量消费时，必须使用 List 接收，否则会抛异常，反之如果不是配置的批量消费，则监听器不能使用 list 接收，否则也会异常。

/**
     * 批量消费时，必须使用 List 接收，否则会抛异常。
     * 即如果配置文件配置的是批量消费(spring.kafka.listener.type=batch)，则监听时必须使用 list 接收
     *
     * @param records
     */
    @KafkaListener(topics = "batch-msg")
    public void messageListener1(List<ConsumerRecord<?, ?>> records) {
        System.out.println(">>>批量消费返回条数，records.size()=" + records.size());
        int count = 0;
        for (ConsumerRecord<?, ?> record : records) {
            System.out.println("\t消息" + (++count) + "：" + record.value());
        }
    }

在线演示源码：https://gitee.com/wangmx1993/apache-kafka/blob/master/src/main/java/com/wmx/apachekafka/controller/KafkaConsumer.java

消费者消费消息并转发消息

1、有的时候，接收消息处理之后，可能需要将其再发送给其它主题，此时当然使用 API 操作也是可以的，不过有更简单的方式，就是使用一个 @SendTo 注解。

2、SendTo 可以标注在类上，此时对类中的所有方法有效，也可以直接标注在方法上，方法的返回值表示转发的消息内容，注解的 value 值表示转发的目标主题。

/**
     * 消费消息并转换。SendTo 可以标注在类上，此时对类中的所有方法有效，方法的返回值表示转发的消息内容。
     *
     * @param record
     * @return
     */
    @KafkaListener(topics = {"sendTo"})
    @SendTo("car-infos")
    public String messageListener3(ConsumerRecord<?, ?> record) {
        System.out.println("消费单条消费并转发：" + record.value() + "," + record.timestamp());
        return record.value().toString();
    }

在线演示源码：https://gitee.com/wangmx1993/apache-kafka/blob/master/src/main/java/com/wmx/apachekafka/controller/KafkaConsumer.java

官网介绍文档：https://docs.spring.io/spring-kafka/docs/2.6.0-SNAPSHOT/reference/html/#annotation-send-to

消费者组（Consumer Group）测试

1、为了进一步提高吞吐量，Kafka 引入了消费者组的概念，由一个或者多个消费者组成一个组，相当于是消费者的集群。

2、比如消费者 A 是一个应用 A 实例，为了提高消费的吞吐量，可以多部署了几个消费者 A 实例（多实例），这样就有多个消费者形成一个消费组，但干的都是应用 A 做的事，需要与消费者B（不同的应用）区分开。

3、一般设置消费组的消费者数量与分区数（partitions）一致，这样一个消费者能负责一个分区，提高效率。如果消费组的消费者数量小于分区数，则会出现一个消费者负责多个分区。而如果消费组的消费者数量大于分区数，则会出现有消费者分不到分区，造成浪费。所以一般保持一致。

4、下面启动两个实例，如果通过 kafka 自身的生产者脚本发送消息，看消费者组内是否只有一个消费成功，演示如下。

Spring.kafka 配置属性介绍

1、Spring boot 对 kafka 的配置以 spring.kafka.* 开头，官网提供了大量的配置选项，通常情况下，绝大部分选项直接使用默认值即可，下面将一些可能常用到的汇总如下。

KafkaTemplate API 介绍

1、KafkaTemplate 包装了一个 producer（生产者），并提供了向 Kafka 主题发送数据的方快捷方法，下表显示了 KafkaTemplate 中的相关方法：

ListenableFuture<SendResult<K, V>> sendDefault(V data);
ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
ListenableFuture<SendResult<K, V>> send(Message<?> message);
Map<MetricName, ? extends Metric> metrics();
List<PartitionInfo> partitionsFor(String topic);
<T> T execute(ProducerCallback<K, V, T> callback);

2、sendDefault API 要求已为模板配置默认主题，execute 方法提供对底层生产者的直接访问。