一．基础概念

1. Kafka概述

Apache Kafka是一个分布式消息发布订阅系统。它最初由LinkedIn公司基于独特的设计实现为一个分布式的提交日志系统( a distributed commit log)，之后成为Apache项目的一部分。Kafka采用订阅-发布模式提供消息服务，支持不同消费组分别消费。在性能方面，它具有高吞吐量、低时延的特点。与Flink、Storm等计算框架兼容性好。在Kafka分布式消息集群中，没有“中心主节点”的概念，集群中所有的服务器都是对等的，因此，可以在不做任何配置的更改的情况下实现服务器的的添加与删除。

2. 消息队列

消息（Message）是指在应用之间传送的数据，消息队列（Message Queue）是一种应用间的通信方式，消息发送后可以立即返回发布者和使用者都不需要感知对方的存在。消息队列系统负责将数据从一个应用程序传输到另一个应用程序，因此应用程序可以专注于数据，但不必担心如何共享数据。 消息队列协同的实现方式有两种类型，一种是点对点，另一种是发布-订阅(pub-sub)。

2.1 消息队列分类

1）点对点消息队列系统

点对点消息队列系统如下图所示，消息被保存在一个队列中。 一个或多个消费者可以消费队列中的消息，但是特定的消息只能由最多一个消费者消费。 一旦消费者在队列中读取消息，消息就从该队列中消失。典型例子是一个订单处理系统，其中每个订单将由一个订单处理器处理，但是多订单处理器也可以同时工作。

2）发布-订阅消息系统

与点对点消息队列系统不同，在发布-订阅消息队列系统中，消息被保存在一个主题中，消费者可以订阅一个或多个主题并使用该主题中的所有消息。在发布-订阅系统中，消息生产者称为发布者，一个发布者可以同时发布多个不同主题的消息；消息消费者称为订阅者，订阅者也可以同时订阅多个主题的消息，分别进行消息读取。

2.2 消息队列对比

1）ActiveMQ

最老牌的开源消息队列。目前已经进入老年期，社区不活跃。无论是功能还是性能方面，都与现代的消息队列存在明显的差距。

2）RabbitMQ

使用Erlang 语言编写，是一个相当轻量级的消息队列。开箱即用易于维护，时延低，但吞吐量较差。

3）RocketMQ

阿里巴巴开源后贡献给Apache，有中文社区。低时延、稳定性高。

4）Kafka

海量的日志系统。采用订阅/发布模式，支持不同消费组分别消费。吞吐量高，时延一般。与Flink、Storm等计算框架兼容性好。

5）Pulsar

与kafka类似，但支持订阅共享，支持持久化存储，支持Exactly once消费。

3. Kafka架构

Kafka功能架构如下图所示：

1）Producer：消息生产者，是消息的产生的源头，负责生成消息并发送到Kafka服务器上。

2）Consumer：消息消费者，是消息的使用方，负责消费Kafka服务器上的消息。

3）Broker：即Kafka的服务器，用户存储消息，Kafa集群中的一台或多台服务器统称为 broker。

4）Group，用于归组同类消费者，在Kafka中，多个消费者可以共同消息一个Topic下的消息，每个消费者消费其中的部分消息，这些消费者就组成了一个分组，拥有同一个分组名称，通常也被称为消费者集群。

5）Offset：消费者拉取消息数据的过程中需要知道消息在文件中的偏移量。

6）ZooKeeper：用于管理和协调Kafka代理。 ZooKeeper服务主要用于通知生产者和消费者系统中存在任何新代理或代理失败。 根据Zookeeper接收到关于代理的存在或失败的通知，然后生产者和消费者采取决定并开始与某些其他代理协调他们的任务。

4. Kafka生产与消费过程

Kafka生产与消费过程如下图所示：

1）Topic：由用户定义并配置在Kafka服务器，用于建立生产者和消息者之间的订阅关系：生产者发送消息到指定的Topic下，消息者从这个Topic下消费消息。可以通过设置Topic的副本数来保证消息的可靠性。

2）Partition：消息主题分区。一个Topic可以分为多个分区。

例如：kafka-test”这个Topic下可以分为6个分区，分别由两台服务器提供，那么通常可以配置为让每台服务器提供3个分区，假如服务器ID分别为0、1，则所有的分区为0-0、0-1、0-2和1-0、1-1、1-2。

可以通过将一个topic切分多任意多个partition的方式来提高消息保存和消费的效率。越多的partitions意味着可以容纳更多的consumer，有效提升并发消费的能力。但在有副本的条件下，越多的分区也意味着需要占用更多的存储空间。

分区的消费模式如下图所示：

一个Topic中的每个partition只会被一个Group中的一个consumer消费，但一个 consumer可以消费多个partitions中的消息（消费者数据小于Partions的数量时），每个group中consumer消息消费互相独立。

注意：kafka的设计原理决定，对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息。

5. Kafka数据传输过程

5.1 数据发送确认

为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 中的每个 partition 收到 producer 发送的数据后，都需要向 producer 发送 ack （acknowledgement 确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。

5.2 消费位移确认

Kafka消费者消费位移确认有自动提交与手动提交两种策略。在创建KafkaConsumer对象时，通过参数enable.auto.commit设定，true表示自动提交（默认）。自动提交策略由消费者协调器（ConsumerCoordinator）每隔${auto.commit.interval.ms}毫秒执行一次偏移量的提交。手动提交需要由客户端自己控制偏移量的提交。

1）自动提交

在创建一个消费者时，默认是自动提交偏移量，当然我们也可以显示设置为自动。例如，我们创建一个消费者，该消费者自动提交偏移量

2）手动提交

在有些场景我们可能对消费偏移量有更精确的管理，以保证消息不被重复消费以及消息不被丢失。假设我们对拉取到的消息需要进行写入数据库处理，或者用于其他网络访问请求等等复杂的业务处理，在这种场景下，所有的业务处理完成后才认为消息被成功消费，这种场景下，我们必须手动控制偏移量的提交。

5.3 数据传输事务性

1）at most once

消息最多发送一次。这个和JMS中"非持久化"消息类似.发送一次，无论成败，将不会重发。at most once:消费者fetch消息,然后保存offset，然后处理消息;当client保存offset之后，但是在消息处理过程中出现了异常，导致部分消息未能继续处理.那么此后"未处理"的消息将不能被fetch到，这就是"atmost once"。

2）at least once

消息至少发送一次，如果消息未能接受成功，可能会重发，直到接收成功。消费者fetch消息，然后处理消息，然后保存offset.如果消息处理成功之后，但是在保存offset阶段zookeeper异常导致保存操作未能执行成功，这就导致接下来再次fetch时可能获得上次已经处理过的消息，这就是"atleast once"，原因offset没有及时的提交给zookeeper，zookeeper恢复正常还是之前offset状态。

3）exactly once

消息只会发送一次。kafka中并没有严格的去实现(基于2阶段提交，事务)，我们认为这种策略在kafka中是没有必要的。通常情况下"at-least-once"是我们首选。相比at most once而言，重复接收数据总比丢失数据要好。

6. Kafka数据存储

消息在Kafka Broker中通Log追加的方式进行持久化存储。并进行分区。为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个数(或尺寸)达到一定阀值时,再flush到磁盘,这样减少了磁盘IO调用的次数。

Kafka Broker中不保存订阅者的状态，由订阅者自己保存。Kafka 采用基于时间的SLA(服务水平保证)，消息保存一定时间（通常为7天）后会被删除。

6.1 Kafka的持久化

1）一个Topic可以认为是一类消息，每个topic将被分成多partition(区),每个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），partition是以文件的形式存储在文件系统中。

2）Logs文件根据broker中的配置要求,保留一定时间后删除来释放磁盘空间。每一个partion(文件夹)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件里但每一个段segment file消息数量不一定相等，这样的特性方便old segment file高速被删除。（默认情况下每一个文件大小为1G）。每一个partiton仅仅须要支持顺序读写即可了。segment文件生命周期由服务端配置參数决定。

3）为数据文件建索引：稀疏存储，每隔一定字节的数据建立一条索引。

下图为一个partition的索引示意图：

6.2 数据分片

由于生产者生产的消息不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。每个 segment 对应两个文件，“.index” 文件和 “.log 文件”。这些文件位于一个文件夹下，该文件夹命名规则为：topic 名称 + 分区序号。

比如创建一个名为firstTopic的topic，其中有3个partition，那么在 kafka 的数据目录（/tmp/kafka-log）中就有 3 个目录，firstTopic-0~3

多个分区在集群中多个broker上的分配方法

1）将所有 N Broker 和待分配的 i 个 Partition 排序

2）将第 i 个 Partition 分配到第(i mod n)个 Broker 上

6.3 日志的清除策略以及压缩策

1）日志的清理策略

a）根据消息的保留时间

当消息在 kafka 中保存的时间超过了指定的时间，就会触发清理过程

b）根据 topic 存储的数据大小

当 topic 所占的日志文件大小大于一定的阀值，则可以开始删除最旧的消息。通过 log.retention.bytes 和 log.retention.hours 这两个参数来设置，当其中任意一个达到要求，都会执行删除。

注：kafka会启动一个后台线程，定期检查是否存在可以删除的消息。

2）日志压缩策略

通过这个功能可以有效的减少日志文件的大小，缓解磁盘紧张的情况，在很多实际场景中，消息的 key 和 value 的值之间的对应关系是不断变化的，就像数据库中的数据会不断被修改一样，消费者只关心 key 对应的最新的 value。因此，我们可以开启 kafka 的日志压缩功能，服务端会在后台启动Cleaner线程池，定期将相同的key进行合并，只保留最新的 value 值。

6.4 Kafka 高效读取数据

1）kafka本身是分布式集群，同时采用分区技术，并发度高。

2）顺序写磁盘，kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100k/s。

3）零复制技术

零拷贝是文件只需要经过Page Cache就可以直接发送出去了，这样就极大的增加了发送数据的效率。应用Page Cache，kafka将数据直接持久化到Page Cache中，其实就是内存中，I/O Scheduler 可以将多个小块的写组装成大块的写操作，降低了I/O次数。

二．安装部署

1. 软件版本

线上版本：

zookeeper: zookeeper-3.7.1

kakfa: kafka_ 2.12-3.0.0

2. Kafka安装

安装JDK1.84. 安装部署

说明：服务都已安装好，确认服务器自启动是否正常

2.1 zookeeper安装

1）下载相应版本，解压即可

https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper

2）配置

vi conf/zoo.cfg

参数说明：

tickTime=2000客户端与服务器或者服务器与服务器之间维持心跳时间

initLimit=10集群中的follower服务器(F)与leader服务器(L)之间初始连接时能容忍的最多心跳数

syncLimit=5集群中flower服务器（F）跟leader（L）服务器之间的请求和答应最多能容忍的心跳数

dataDir存放myid信息跟一些版本，日志，跟服务器唯一的ID信息

dataLogDir日志存放路径，（需修改启动脚本）clientPort服务端口server.1集群信息

3）zookeeper启停

停止： ./bin/zkServer.sh stop

启动： ./bin/zkServer.sh start

2.2 kafka安装

1）下载相应版本，解压即可

http://kafka.apache.org/downloads

2）配置

vi config/server.properties

参数说明：

broker.id =0每一个broker在集群中的唯一表示，要求是正数

log.dirs=/data/kafka-logskafka数据的存放地址

port =9092broker server服务端口

message.max.bytes =6525000表示消息体的最大大小，单位是字节

num.network.threads =4broker处理消息的最大线程数，一般情况下数量为cpu核数

num.io.threads =8broker处理磁盘IO的线程数，数值为cpu核数2倍

background.threads =4一些后台任务处理的线程数，例如过期消息文件的删除等，一般情况下不需要去做修改

queued.max.requests =500等待IO线程处理的请求队列最大数，若是等待IO的请求超过这个数值，那么会停止接受外部消息，应该是一种自我保护机制host.namebroker的主机地址socket.send.buffer.bytes=100*1024socket的发送缓冲区，

socket的调优参数SO_SNDBUFF

socket.receive.buffer.bytes =100*1024socket的接受缓冲区，

socket的调优参数SO_RCVBUFF

socket.request.max.bytes =10010241024socket请求的最大数值，防止serverOOM，message.max.bytes必然要小于socket.request.max.bytes，会被topic创建时的指定参数覆盖delete.topic.enable=true是否彻底删除Topic5. 启停方法

3）kafka启停

停止： ../bin/kafka-server-stop.sh

启动： ./bin/kafka-server-start.sh -daemon config/server.properties6. 集群配置

3. Kafka集群配置

1）配置zookeeper集群(3台为例)

a）vi conf/zoo.cfg

添加集群机器信息

server.1=172.26.1.247:2181

server.2=172.26.1.247:2182

server.3=172.26.1.247:2183

b）每台机器上生成响应的id文件(不能重复)

server1上执行： echo "1" > /home/appadmin/kafka/data/myid

server2上执行： echo "2" > /home/appadmin/kafka/data/myid

server3上执行： echo "3" > /home/appadmin/kafka/data/myid

c）重启zookeeper服务

停止 ./bin/zkServer.sh stop

启动：./bin/zkServer.sh start

检查集群状态：./bin/zkServer.sh status 可以查看节点的模式，分为follower与leader

2）配置kafka集群

a）vi config/server.properties

添加zookeeper集群机器信息

zookeeper.connect=172.26.1.247:2181, 172.26.1.247:2182, 172.26.1.247:2183

zookeeper.connection.timeout.ms=400000

修改broker.id(每台机器的id不能重复)

broker.id=20

b）重启kafka服务

停止 bin/kafka-server-stop.sh

启动 bin/kafka-server-start.sh -daemon config/server.properties

c）测试集群

消费者服务器查看到数据则集群配置完成

3）配置优化

num.io.threads=8 --broker io处理的线程数，这个数量一定要比log.dirs的目录数要大

The send buffer (SO_SNDBUF) used by the socket server

socket.send.buffer.bytes=1048576 --将发送的消息先放到缓冲区，当到达一定量的时候再一次性发出

The receive buffer (SO_RCVBUF) used by the socket server

socket.receive.buffer.bytes=1048576 --kafka接受消息的缓冲区，当接受的数量达到一定量的时候再写入磁盘

为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个数(或尺寸)达到一定阀值时,再flush到磁盘,这样减少了磁盘IO调用的次数。

The maximum size of a request that the socket server will accept (protection against OOM)

socket.request.max.bytes=104857600 --像kafka发送或者请求消息的最大数，此设置不能超过java堆栈大小

4. Kafka常用命令

1）创建一个topic

sh kafka-topics.sh --create --topic tyjt-test1 --bootstrap-server 172.26.1.246:9092 --replication-factor 1 --partitions 1

2）查看topic

sh kafka-topics.sh --list --bootstrap-server 172.26.1.246:9092 --topic tyjt-test

3）查看topic

bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic test

4）删除topic

sh kafka-topics.sh --delete --bootstrap-server 172.26.1.246:9092 --topic tyjt-test1

删除topic：慎用，只会删除zookeeper中的元数据，消息文件须手动删除

5）列出所有主题中的所有用户组

./kafka-consumer-groups.sh --bootstrap-server 172.26.1.246:9092 –list

要使用ConsumerOffsetChecker查看上一个示例中消费者组的偏移量

./kafka-consumer-groups.sh --bootstrap-server 172.26.1.246:9092 --describe --group lushijie3

6）修改partitions数量

kafka-topics.sh --bootstrap-server 172.26.1.246:9092 --topic tyjt-test --alter --partitions 4

7）模拟生产者

sh kafka-console-producer.sh --broker-list 172.26.1.246:9092 --topic tyjt-test

8）模拟消费者

sh kafka-console-consumer.sh --bootstrap-server 172.26.1.246:9092 --topic tyjt-test

可以看到当前的消费进度（CURRENT-OFFSET）、消息进度（LOG-END-OFFSET）、落后量（LAG）:

./kafka-consumer-groups.sh --bootstrap-server 172.26.1.246:9092 --describe --group lushijie3

三．Kafka-sdk接入

1. 原理介绍

重写spring-kafka中的两个关键的bean对象，ProducerFactory（kafka生产者工厂对象）和KafkaListenerContainerFactory（kafka消费者工厂对象），从而实现个性化配置；使用者无需关注具体配置，只要在配置文件中开启sdk的使用即可

2. 接入流程

1）POM文件引入SDK

<dependency>
<groupId>com.tyjt</groupId>
<artifactId>ccp-toolkit-mq</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>

2） application.yml配置文件

kafka:
    producer:
        used: true
    consumer:
        used: true
3）bootstrap.yml配置文件

spring:
    cloud:
        nacos:
        username: nacos
        password: nacos
        config:
            server-addr: 172.26.1.247:8848
            namespace: 1d46279c-0cea-42bf-a976-cd31f8170ef2
            file-extension: yml
            name: ccp-mq

4）代码示例

参考 ccp-toolkit-example/ccp-toolkit-mq-example

生产者示例代码ProducerService.java

@Service
    @Slf4j
    public class ProducerService {

        @Resource
        private KafkaTemplate<String, Object> kafkaTemplate;

        @Scheduled(fixedRate = 10 * 1000)
        public void sendMsg() {
            log.info("开始发送消息 kafka");
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("tyjt-test",
                                                                                     "data", "kafka_message_" + System.currentTimeMillis() / 1000);
            future.addCallback(new ListenableFutureCallback<>() {
                @Override
                public void onFailure(Throwable ex) {
                    log.error("发送消息 kafka 失败", ex);
                }

                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    log.info("发送消息 kafka 成功");
                }
            });
        }

    }

消费者示例代码ConsumerService.java

@Service
    @Slf4j
    public class ConsumerService {

        @KafkaListener(topics = "tyjt-test", groupId = "lushijie")
        public void listen(List<ConsumerRecord<String, Object>> recordList) {
            recordList.forEach(data -> 
                               log.info("获取kafka数据,topic:{},data:{},分区:{},偏移量:{}", data.topic(), data.value(),
                                                data.partition(), data.offset()));
        }

    }

3. 验证方式

1）工厂初始化日志

2）数据日志