Java人应该知道的SpringBoot For Kafka (上)

点击上方☝SpringForAll社区 轻松关注！

及时获取有趣有料的技术文章

本文来源：http://www.mydlq.club/article/34/

. 一、概念知识

. 什么是消息中间件

. 什么是 Kafka

. Kafka 特性

. 使用场景

. 基本概念

. 生产者 ACKS 机制

. 消费者更新 Offset 偏移量两种方式

. 二、SpringBoot 操作 Kafka 示例

. 1、Maven 引入 Kafka 相关组件

. 2、Topic 配置

. 3、Producer 配置

. 4、Consumer 配置

---

参考信息：

• kafka 官方网址

• spring-kafka 2.2.7 版本文档

• 示例 Github 地址：https://github.com/my-dlq/blog-example/tree/master/springboot/springboot-kafka

环境说明：

• Kafka 版本：2.3.0

• Zookeeper 版本：3.4.14

• SpringBoot 版本：2.1.7.RELEASE

• Spring For Apache Kafka 版本：2.2.8

一、概念知识

什么是消息中间件

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。

什么是 Kafka

Apache Kafka 是一个分布式高吞吐量的流消息系统，Kafka 建立在 ZooKeeper 同步服务之上。它与 Apache Storm 和 Spark 完美集成，用于实时流数据分析，与其他消息传递系统相比，Kafka具有更好的吞吐量，内置分区，数据副本和高度容错功能，因此非常适合大型消息处理应用场景。

Kafka 特性

• 高并发： 支持数千个客户端同时读写。

• 可扩展性： kafka集群支持热扩展。

• 容错性： 允许集群中节点失败(若副本数量为n，则允许n-1个节点失败)。

• 持久性、可靠性： 消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。

• 高吞吐量、低延迟： Kafka每秒可以处理几十万消息，延迟最低只有几毫秒，每个消息主题topic可以分多个区，消费者组(consumer group)对消息分区(partition)进行消费。

使用场景

• 日志收集： 可以用 kafka 收集各种服务的日志，通过kafka以统一接口服务的方式开放给各种消费者，如 hadoop，Hbase，Solr 等。

• 消息系统： 解耦生产者和消费者、缓存消息等。

• 用户活动跟踪： Kafka 经常被用来记录web用户或者app用户的各种活动，如浏览网页，搜索，点击等活动，这些活动信息被各个服务器发布到 kafka 的 topic 中，然后订阅者通过订阅这些 topic 来做实时的监控分析，或者装载到 hadoop、数据仓库中做离线分析和挖掘。

• 运营指标： Kafka也经常用来记录运营监控数据，包括收集各种分布式应用的数据，比如报警和报告等。

• 流式处理： 比如 spark streaming 和 storm。

基本概念

• Broker： 消息中间件处理节点，一个 Kafka 节点就是一个 Broker，一个或者多个 Broker 可以组成一个 Kafka 集群。

• Topic： Kafka 的消息通过 Topic 主题来分类，Topic类似于关系型数据库中的表，每个 Topic 包含一个或多（Partition）分区。

• Partition： 多个分区会分布在Kafka集群的不同服务节点上，消息以追加的方式写入一个或多个分区中。

• LogSegment： 每个分区又被划分为多个日志分段 LogSegment 组成，日志段是 Kafka 日志对象分片的最小单位。LogSegment 算是一个逻辑概念，对应一个具体的日志文件（”.log” 的数据文件）和两个索引文件（”.index” 和 “.timeindex”，分别表示偏移量索引文件和消息时间戳索引文件）组成。

• Offset： 每个分区中都由一系列有序的、不可变的消息组成，这些消息被顺序地追加到 Partition 中，每个消息都有一个连续的序列号称之为 Offset 偏移量，用于在 Partition 内唯一标识消息。

• Message： 消息是 Kafka 中存储的最小最基本的单位，即为一个 commit log，由一个固定长度的消息头和一个可变长度的消息体组成。

• Producer： 消息的生产者，负责发布消息到 Kafka Broker，生产者在默认情况下把消息均衡地分布到主题的所有分区上，用户也可以自定义分区器来实现消息的分区路由。

• Consumer： 消息的消费者，从 Kafka Broker 读取消息的客户端，消费者把每个分区最后读取的消息的 Offset 偏移量保存在 Zookeeper 或 Kafka 上，如果消费者关闭或重启，它的读取状态不会丢失。

• Consumer Group： 每个 Consumer 属于一个特定的 Consumer Group（若不指定 Group Name则属于默认的 group），一个或多个 Consumer 组成的群组可以共同消费一个 Topic 中的消息，但每个分区只能被群组中的一个消费者操作。

生产者 ACKS 机制

ACKS 参数指定了必须要有多少个分区副本接收到消息，生产者才会认为消息写入是发送消息成功的，这个参数对消息丢失的可能性会产生重要影响，主参数有如下选项：

• acks=0： 把消息发送到kafka就认为发送成功。

• acks=1： 把消息发送到kafka leader分区，并且写入磁盘就认为发送成功。

• acks=all： 把消息发送到 Kafka Leader 分区，并且 Leader 分区的副本 Follower 对消息进行了同步就认为发送成功。

消费者更新 Offset 偏移量两种方式

详情可以查看参考的一篇文章：https://www.jianshu.com/p/d5cd34e429a2

消费者把每个分区最后读取的悄息偏移量提交保存在 Zookeeper 或 Kafka 上，如果消费者关闭或重启，它的读取状态不会丢失，KafkaConsumer API 提供了很多种方式来提交偏移量，但是不同的提交方式会产生不同的数据影响。

• 自动提交：

如果 enable.auto.commit 被设置为 true，那么消费者会自动提交当前处理到的偏移量存入 Zookeeper，自动提交的时间间隔为5s，通过 atuo.commit.interval.ms 属性设置，自动提交是非常方便，但是自动提交会出现消息被重复消费的风险，可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复悄息的时间窗，不过这种情况是无也完全避免的。

• 手动提交：

鉴于 Kafka 自动提交 Offset 的不灵活性和不精确性（只能是按指定频率的提交），Kafka提供了手动提交 Offset 策略，将 auto.commit.offset 自动提交参数设置为 false 来关闭自动提交开启手动模式，手动提交能对偏移量更加灵活精准地控制，以保证消息不被重复消费以及消息不被丢失。

二、SpringBoot 操作 Kafka 示例

1、Maven 引入 Kafka 相关组件

 1<?xml version="1.0" encoding="UTF-8"?>
 2<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 3         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 4    <modelVersion>4.0.0</modelVersion>
 5    <parent>
 6        <groupId>org.springframework.boot</groupId>
 7        <artifactId>spring-boot-starter-parent</artifactId>
 8        <version>2.1.7.RELEASE</version>
 9    </parent>
10
11    <groupId>club.mydlq</groupId>
12    <artifactId>springboot-kafka-demo</artifactId>
13    <version>0.0.1-SNAPSHOT</version>
14    <name>springboot-kafka-demo</name>
15
16    <properties>
17        <java.version>1.8</java.version>
18    </properties>
19
20    <dependencies>
21        <dependency>
22            <groupId>org.springframework.boot</groupId>
23            <artifactId>spring-boot-starter-web</artifactId>
24        </dependency>
25        <dependency>
26            <groupId>org.springframework.kafka</groupId>
27            <artifactId>spring-kafka</artifactId>
28        </dependency>
29    </dependencies>
30
31    <build>
32        <plugins>
33            <plugin>
34                <groupId>org.springframework.boot</groupId>
35                <artifactId>spring-boot-maven-plugin</artifactId>
36            </plugin>
37        </plugins>
38    </build>
39
40</project>

2、Topic 配置

配置 Topic，每次程序启动时检测 Kafka 中是否存在已经配置的 Topic，如果不存在就创建。

 1import org.apache.kafka.clients.admin.AdminClientConfig;
 2import org.apache.kafka.clients.admin.NewTopic;
 3import org.springframework.context.annotation.Bean;
 4import org.springframework.context.annotation.Configuration;
 5import org.springframework.kafka.core.KafkaAdmin;
 6import java.util.HashMap;
 7import java.util.Map;
 8
 9@Configuration
10public class KafkaTopicConfig {
11
12    /**
13     * 定义一个KafkaAdmin的bean，可以自动检测集群中是否存在topic，不存在则创建
14     */
15    @Bean
16    public KafkaAdmin kafkaAdmin() {
17        Map<String, Object> configs = new HashMap<>();
18        // 指定多个kafka集群多个地址，例如：192.168.2.11,9092,192.168.2.12:9092,192.168.2.13:9092
19        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
20        return new KafkaAdmin(configs);
21    }
22
23    /**
24     * 创建 Topic
25     */
26    @Bean
27    public NewTopic topicinfo() {
28        // 创建topic，需要指定创建的topic的"名称"、"分区数"、"副本数量(副本数数目设置要小于Broker数量)"
29        return new NewTopic("test", 3, (short) 0);
30    }
31
32}

3、Producer 配置

(1)、创建 Producer 配置类

创建 Producer 配置类，对 Kafka 生产者进行配置,在配置中需要设置三个 Bean 分别为：

• kafkaTemplate：kafka template 实例，用于 Spring 中的其它对象引入该 Bean，通过其向 Kafka 发送消息。

• producerFactory：producer 工厂，用于对 kafka producer 进行配置。

• producerConfigs：对 kafka producer 参数进行配置。

 1import org.apache.kafka.clients.producer.ProducerConfig;
 2import org.apache.kafka.common.serialization.StringSerializer;
 3import org.springframework.context.annotation.Bean;
 4import org.springframework.context.annotation.Configuration;
 5import org.springframework.kafka.annotation.EnableKafka;
 6import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 7import org.springframework.kafka.core.KafkaTemplate;
 8import org.springframework.kafka.core.ProducerFactory;
 9import java.util.HashMap;
10import java.util.Map;
11
12// 设置@Configuration、@EnableKafka两个注解，声明Config并且打开KafkaTemplate能力。
13@Configuration
14@EnableKafka
15public class KafkaProducerConfig {
16
17    /**
18     * Producer Template 配置
19     */
20    @Bean(name="kafkaTemplate")
21    public KafkaTemplate<String, String> kafkaTemplate() {
22        return new KafkaTemplate<>(producerFactory());
23    }
24
25    /**
26     * Producer 工厂配置
27     */
28    public ProducerFactory<String, String> producerFactory() {
29        return new DefaultKafkaProducerFactory<>(producerConfigs());
30    }
31
32    /**
33     * Producer 参数配置
34     */
35    public Map<String, Object> producerConfigs() {
36        Map<String, Object> props = new HashMap<>();
37        // 指定多个kafka集群多个地址
38        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
39        // 重试次数，0为不启用重试机制
40        props.put(ProducerConfig.RETRIES_CONFIG, 0);
41        // acks=0 把消息发送到kafka就认为发送成功
42        // acks=1 把消息发送到kafka leader分区，并且写入磁盘就认为发送成功
43        // acks=all 把消息发送到kafka leader分区，并且leader分区的副本follower对消息进行了同步就任务发送成功
44        props.put(ProducerConfig.ACKS_CONFIG,"1");
45        // 生产者空间不足时，send()被阻塞的时间，默认60s
46        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 6000);
47        // 控制批处理大小，单位为字节
48        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
49        // 批量发送，延迟为1毫秒，启用该功能能有效减少生产者发送消息次数，从而提高并发量
50        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
51        // 生产者可以使用的总内存字节来缓冲等待发送到服务器的记录
52        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 40960);
53        // 消息的最大大小限制,也就是说send的消息大小不能超过这个限制, 默认1048576(1MB)
54        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,1048576);
55        // 键的序列化方式
56        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
57        // 值的序列化方式
58        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
59        // 压缩消息，支持四种类型，分别为：none、lz4、gzip、snappy，默认为none。
60        // 消费者默认支持解压，所以压缩设置在生产者，消费者无需设置。
61        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"none");
62        return props;
63    }
64
65}

(2)、创建 Producer Service 向 kafka 发送数据

创建 Producer Service 引入 KafkaTemplate 对象，再创建 sendMessageSync、sendMessageAsync 两个方法，分别利用“同步/异步”两种方法向 kafka 发送消息。

 1import org.springframework.beans.factory.annotation.Autowired;
 2import org.springframework.kafka.core.KafkaTemplate;
 3import org.springframework.kafka.support.SendResult;
 4import org.springframework.stereotype.Service;
 5import org.springframework.util.concurrent.ListenableFuture;
 6import org.springframework.util.concurrent.ListenableFutureCallback;
 7import java.util.concurrent.ExecutionException;
 8import java.util.concurrent.TimeUnit;
 9import java.util.concurrent.TimeoutException;
10
11@Service
12public class KafkaProducerService {
13
14    @Autowired
15    private KafkaTemplate kafkaTemplate;
16
17    /**
18     * producer 同步方式发送数据
19     * @param topic    topic名称
20     * @param message  producer发送的数据
21     */
22    public void sendMessageSync(String topic, String message) throws InterruptedException, ExecutionException, TimeoutException {
23        kafkaTemplate.send(topic, message).get(10, TimeUnit.SECONDS);
24    }
25
26    /**
27     * producer 异步方式发送数据
28     * @param topic    topic名称
29     * @param message  producer发送的数据
30     */
31    public void sendMessageAsync(String topic, String message) {
32        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic, message);
33        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
34            @Override
35            public void onSuccess(SendResult<Integer, String> result) {
36                System.out.println("success");
37            }
38
39            @Override
40            public void onFailure(Throwable ex) {
41                System.out.println("failure");
42            }
43        });
44    }
45
46}

(3)、创建 Producer Controller 调用 Producer Service 产生数据

Spring Controller 类，用于调用 Producer Service 中的方法向 kafka 发送消息。

 1import club.mydlq.springbootkafkademo.service.ProducerService;
 2import org.springframework.beans.factory.annotation.Autowired;
 3import org.springframework.web.bind.annotation.GetMapping;
 4import org.springframework.web.bind.annotation.RestController;
 5import java.util.concurrent.ExecutionException;
 6import java.util.concurrent.TimeoutException;
 7
 8@RestController
 9public class KafkaProducerController {
10
11    @Autowired
12    private KafkaProducerService producerService;
13
14    @GetMapping("/sync")
15    public void sendMessageSync() throws InterruptedException, ExecutionException, TimeoutException {
16        producerService.sendMessageSync("test","同步发送消息测试");
17    }
18
19    @GetMapping("/async")
20    public void sendMessageAsync(){
21        producerService.sendMessageAsync("test","异步发送消息测试");
22    }
23
24}

4、Consumer 配置

(1)、创建 Consumer 配置类

创建 Consumer 配置类，对 Kafka 消费者进行配置,在配置中需要设置三个 Bean 分别为：

• kafkaListenerContainerFactory：kafka container 工厂，负责创 建container，当使用@KafkaListener时需要提供。

• consumerFactory：consumer 工厂，用于对 kafka consumer 进行配置。

• consumerConfigs：对 kafka consumer 参数进行配置。

 1import org.apache.kafka.clients.consumer.ConsumerConfig;
 2import org.apache.kafka.common.serialization.StringDeserializer;
 3import org.springframework.context.annotation.Bean;
 4import org.springframework.context.annotation.Configuration;
 5import org.springframework.kafka.annotation.EnableKafka;
 6import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 7import org.springframework.kafka.config.KafkaListenerContainerFactory;
 8import org.springframework.kafka.core.ConsumerFactory;
 9import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
10import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
11import java.util.HashMap;
12import java.util.Map;
13
14@Configuration
15@EnableKafka
16public class KafkaConsumerConfig {
17
18    @Bean
19    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
20        ConcurrentKafkaListenerContainerFactory<Integer, String>
21                factory = new ConcurrentKafkaListenerContainerFactory<>();
22        // 设置消费者工厂
23        factory.setConsumerFactory(consumerFactory());
24        // 消费者组中线程数量
25        factory.setConcurrency(3);
26        // 拉取超时时间
27        factory.getContainerProperties().setPollTimeout(3000);
28        return factory;
29    }
30
31    @Bean
32    public ConsumerFactory<Integer, String> consumerFactory() {
33        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
34    }
35
36    @Bean
37    public Map<String, Object> consumerConfigs() {
38        Map<String, Object> propsMap = new HashMap<>();
39        // Kafka地址
40        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
41        // 是否自动提交offset偏移量(默认true)
42        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
43        // 自动提交的频率(ms)
44        propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
45        // Session超时设置
46        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
47        // 键的反序列化方式
48        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
49        // 值的反序列化方式
50        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
51        // offset偏移量规则设置：
52        // (1)、earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
53        // (2)、latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
54        // (3)、none：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
55        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
56        return propsMap;
57    }
58
59}

(2)、创建 Consumer Service 监听 Kafka 数据

 1import org.springframework.kafka.annotation.KafkaListener;
 2import org.springframework.stereotype.Service;
 3
 4@Service
 5public class KafkaConsumerService {
 6
 7    @KafkaListener(topics = {"test"},groupId = "group1", containerFactory="kafkaListenerContainerFactory")
 8    public void kafkaListener(String message){
 9        System.out.println(message);
10    }
11
12}