【Kafka】-阿里云开发者社区

一、Kafka是什么？

定义：Kafka是一个基于zookeeper协调的分布式、多副本的（replica）、支持分区的（partition）系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写的项目。

二、Kafka的安装与配置

一、Docker安装kafka

Kafka是用Scala语言开发的，运行在JVM上，在安装Kafka之前需要先安装JDK。

yum install java-1.8.0-openjdk* -y

下载zookeeper镜像

docker pull wurstmeister/zookeeper

开放端口

firewall-cmd --add-port=2181/tcp --permanent
firewall-cmd --reload
firewall-cmd --query-port=2181/tcp
systemctl restart docker

启动镜像生成容器

docker run -dit --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2  --name zookeeper \
-p 2181:2181 \
-v /etc/localtime:/etc/localtime \
-t wurstmeister/zookeeper

下载kafka镜像

docker pull wurstmeister/kafka

开放端口

firewall-cmd --add-port=9092/tcp --permanent
firewall-cmd --reload
firewall-cmd --query-port=9092/tcp
systemctl restart docker

启动kafka镜像生成容器

docker run -dit --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 --name kafka \
-p 9092:9092 \
-e KAFKA_BROKER_ID=0 \
-e KAFKA_ZOOKEEPER_CONNECT=172.21.17.47:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.21.17.47:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka

参数说明：

-e KAFKA_BROKER_ID=0 在kafka集群中，每个kafka都有一个BROKER_ID来区分自己
-e KAFKA_ZOOKEEPER_CONNECT=ip:2181/kafka 配置zookeeper管理kafka的路径ip:2181/kafka,ip地址改为内网ip
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip:9092 把kafka的地址端口注册给zookeeper，ip地址改成内网ip
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 配置kafka的监听端口
-v /etc/localtime:/etc/localtime 容器时间同步虚拟机的时间

验证kafka是否可以使用

进入容器

docker exec -it kafka bash

进入 /opt/kafka_2.13-2.7.1/bin/ 目录下

cd /opt/kafka_2.13-2.7.1/bin/

运行kafka生产者发送消息

./kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic sun

发送消息

{"datas[{"channel":"","metric":"temperature","producer":"ijinus","sn":"IJA0101-00002245","time":"1543207156000","value":"80"}],"ver":"1.0"}

运行kafka消费者接收消息

./kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --topic sun --from-beginning

停止zookeeper和kafka

docker stop zookeeper
docker rm zookeeper
docker stop kafka
docker rm kafka

二、安装包安装kafka

一、安装JDK

yum install java-1.8.0-openjdk* -y

二、安装Zookeeper

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz
tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
cd apache-zookeeper-3.5.8-bin/
cp conf/zoo_sample.cfg conf/zoo.cfg
cd /home/apache-zookeeper-3.5.8-bin/bin
./zkServer.sh start
./zkCli.sh
ls /

打印结果：

[zk: localhost:2181(CONNECTED) 5] ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, kafka, latest_producer_id_block, log_dir_event_notification, zookeeper]
[zk: localhost:2181(CONNECTED) 6]

三、安装Kafka

wget https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgz
tar -xzf kafka_2.11-2.4.1.tgz
cd /home/kafka_2.11-2.4.1/config
vim config/server.properties

配置文件编辑

#broker.id属性在kafka集群中必须要是唯一
broker.id=0
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://内网ip:9092   
#kafka的消息存储文件
log.dir=/usr/local/data/kafka-logs
#kafka连接zookeeper的地址
zookeeper.connect=内网ip:2181

如果填写外网ip可能会遇上这种情况：

四、启动并验证kafka

启动kafka

/home/kafka_2.11-2.4.1/bin/kafka-server-start.sh config/server.properties &

进入zookeeper目录通过zookeeper客户端查看下zookeeper的目录树

/home/apache-zookeeper-3.5.8-bin/bin/zkCli.sh
ls /
ls /brokers/ids

校验kafka

创建主题

/home/kafka_2.11-2.4.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 1 --topic test

查看kafka中目前存在的topic

/home/kafka_2.11-2.4.1/bin/kafka-topics.sh --list --zookeeper 106.14.132.94:2181

发送消息

/home/kafka_2.11-2.4.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a msg

消费消息

/home/kafka_2.11-2.4.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092--topic test

消费之前的消息

/home/kafka_2.11-2.4.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --topic test

通过jps命令查看运行的情况

对于kafka来说，一个单独的broker意味着kafka集群中只有一个节点。要想增加kafka集群中的节点数量，只需要多启动几个broker实例即可。为了有更好的理解，现在我们在一台机器上同时启动三个broker实例。

三、集群搭建与使用

一、集群配置

首先，我们需要建立好其他2个broker的配置文件：

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

配置文件的需要修改的内容分别如下：
config/server-1.properties:

#broker.id属性在kafka集群中必须要是唯一
broker.id=1
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://106.14.132.94:9093   
log.dir=/usr/local/data/kafka-logs-1
#kafka连接zookeeper的地址，要把多个kafka实例组成集群，对应连接的zookeeper必须相同
zookeeper.connect=106.14.132.94:2181

config/server-2.properties:

broker.id=2
listeners=PLAINTEXT://106.14.132.94:9094
log.dir=/usr/local/data/kafka-logs-2
zookeeper.connect=106.14.132.94:2181

目前我们已经有一个zookeeper实例和一个broker实例在运行了，现在我们只需要在启动2个broker实例即可：

bin/kafka-server-start.sh -daemon config/server-1.properties
bin/kafka-server-start.sh -daemon config/server-2.properties

二、验证

查看zookeeper确认集群节点是否都注册成功：

ls /brokers/ids

现在我们创建一个新的topic，副本数设置为3，分区数设置为2：

bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic

查看下topic的情况

 bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic my-replicated-topic

以下是输出内容的解释，第一行是所有分区的概要信息，之后的每一行表示每一个partition的信息。
leader节点负责给定partition的所有读写请求，同一个主题不同分区leader副本一般不一样(为了容灾)
replicas 表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

现在我们向新建的 my-replicated-topic 中发送一些message，kafka集群可以加上所有kafka节点：

bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --topic my-replicated-topic

my test msg 1
my test msg 2

现在开始消费：

bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --from-beginning --topic my-replicated-topic

my test msg 1
my test msg 2

现在我们来测试我们容错性，因为broker1目前是my-replicated-topic的分区0的leader，所以我们要将其kill

ps -ef | grep server.properties
kill 14776

现在再执行命令：

bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:9092 --topic my-replicated-topic

我们可以看到，分区0的leader节点已经变成了broker 0。要注意的是，在Isr中，已经没有了1号节点。leader的选举也是从ISR(in-sync replica)中进行的。
此时，我们依然可以消费新消息：

bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --from-beginning --topic my-replicated-topic

my test msg 1
my test msg 2

查看主题分区对应的leader信息：

get /brokers/topics/my-relicated-topic/partitions/0/state

kafka将很多集群关键信息记录在zookeeper里，保证自己的无状态，从而在水平扩容时非常方便。

三、集群消费

log的partitions分布在kafka集群中不同的broker上，每个broker可以请求备份其他broker上partition上的数据。kafka集群支持配置一个partition备份的数量。针对每个partition，都有一个broker起到“leader”的作用，0个或多个其他的broker作为“follwers”的作用。leader处理所有的针对这个partition的读写请求，而followers被动复制leader的结果，不提供读写(主要是为了保证多副本数据与消费的一致性)。如果这个leader失效了，其中的一个follower将会自动的变成新的leader。

Producers
生产者将消息发送到topic中去，同时负责选择将message发送到topic的哪一个partition中。通过round-robin做简单的负载均衡。也可以根据消息中的某一个关键字来进行区分。通常第二种方式使用的更多。

Consumers
传统的消息传递模式有2种：队列( queue) 和（publish-subscribe）

queue模式：多个consumer从服务器中读取数据，消息只会到达一个consumer。
publish-subscribe模式：消息会被广播给所有的consumer。

Kafka基于这2种模式提供了一种consumer的抽象概念：消费组（consumer group）。

queue模式：所有的consumer都位于同一个consumer group 下。
publish-subscribe模式：所有的consumer都有着自己唯一的consumer group。

上图说明：由2个broker组成的kafka集群，某个主题总共有4个partition(P0-P3)，分别位于不同的broker上。这个集群由2个Consumer Group消费， A有2个consumer instances ，B有4个。通常一个topic会有几个consumer group，每个consumer group都是一个逻辑上的订阅者（ logical subscriber ）。每个consumer group由多个consumer instance组成，从而达到可扩展和容灾的功能。

四、Kafka可视化管理工具kafka-manager

安装及基本使用可参考：Java廖志伟

三、Kafka的使用场景

日志收集：一个公司可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
消息系统：解耦和生产者和消费者、缓存消息等。
用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

四、Kafka基本概念

Broker：消息中间件处理节点，一个Kafka节点就是一个broker，一个或者多个Broker可以组成一个Kafka集群。
Topic：Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic。
Producer：消息生产者，向Broker发送消息的客户端。 Consumer：消息消费者，从Broker读取消息的客户端。
ConsumerGroup：每个Consumer属于一个特定的Consumer Group，一条消息可以被多个不同的ConsumerGroup消费，但是一个Consumer Group中只能有一个Consumer能够消费该消息。
Partition：物理上的概念，一个topic可以分为多个partition，每个partition内部消息是有序的。

五、Kafka的初体验

创建主题

创建一个名字为“test”的Topic，这个topic只有一个partition，并且备份因子也设置为1。当producer发布一个消息到某个指定的Topic，这个Topic如果不存在，就自动创建。

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 1 --topic test

查看kafka中目前存在的topic

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --list --zookeeper 106.14.132.94:2181

列表中有一个__consumer_offsets主题，这个主题不能删除哟

删除主题

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --delete --topic test --zookeeper 106.14.132.94:2181

如果出现This will have no impact if delete.topic.enable is not set to true.

彻底删除topic：
[1. ] 删除Topic，delete.topic.enable=true这里要设置为true，需要在$KAFKA/config/server.properties中配置delete.topic.enble=true
[2. ] 删除log日志
[3. ] 删除ZK中的Topic记录

删除列表中有一个consumer_offsets主题会出现，Topic consumer_offsets is a kafka internal topic and is not allowed to be marked for deletion.

__consumer_offsets这个topic是由kafka自动创建的，默认49个，这个topic是不能被删除的！

为什么这里会是这样存储__consumer_offsets的呢？

[1.] 将所有 N Broker 和待分配的 i 个 Partition 排序
[2.] 将第 i 个 Partition 分配到第(i mod n)个 Broker 上

发送消息

kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。首先我们要运行发布消息的脚本，然后在命令中输入要发送的消息的内容。

/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a news
this is a another news

消费消息

对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输出，默认是消费最新的消息。由于我们已经发送了消息了，想要消费之前的消息可以通过--from-beginning参数指定。

消费之前的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --topic test

消费最新的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --topic test

通过不同的终端窗口来运行以上的命令，你将会看到在producer终端输入的内容，很快就会在consumer的终端窗口上显示出来。

消费之前多主题的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --whitelist "test|test2"

六、消费模式

单播消费

一条消息只能被某一个消费者消费的模式，类似queue模式，只需让所有消费者在同一个消费组里即可。

分别在两个客户端执行如下消费命令，然后往主题里发送消息，结果只有一个客户端能收到消息。
发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

多播消费

一条消息能被多个消费者消费的模式，类似publish-subscribe模式费，针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者，该消费者属于testGroup-2消费组，结果两个客户端都能收到消息。

发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a more news
this is a more new2
this is a more new3

客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端3：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup2 --topic test

七、消费组的偏移量（offset）

查看消费组名

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --list

查看消费组的消费偏移量

查看消费组testGroup和testGroup2的消费偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2

消费组的消费偏移量参数

current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数

发送消息理解消费组的消费偏移量

1个客户端的消费组是testGroup，一个客户端的消费组是testGroup，他们二个都是同一个test主题。

发送一条消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a testgroup message

查看偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

当前消费组的已消费偏移量（current-offset）没变，说明当前的消费组没有消费者进行消费，因为没有启动消费者。主题对应分区消息的结束偏移量（log-end-offset）加1了，说明当前分区的消息多了一条。当前消费组未消费的消息数（lag）加1了，说明有1条消息没有被消费。

客户端的消费组testGroup开始消费

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092  --consumer-property group.id=testGroup --topic test

再次查看消费组testGroup和testGroup2的消费偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2

发现testGroup消费组当前消费组的已消费偏移量（current-offset）加1了，当前消费组未消费的消息数（lag）减1了，说明testGroup已经被消费了。testGroup2消费者还是一样没有发生变化，所以可以推测出，消费者是通过不同的消费组进行消费的，每个消费组互不影响。

八、主题/分区/日志的概念

主题

Topic是一个类别的名称，同类消息发送到同一个Topic下面。对于每一个Topic，下面可以有多个分区(Partition)日志文件。Topic是一个逻辑概念，真正落实的数据都在分区上面，一般每个主题至少有一个分区。这样可以通过多个分区，放在在不同的服务器上面去，达到分布式存储的功能，海量数据通过分区存储，切片存储解决单台机器无法存储海量数据的问题。

分区

分区是一个有序的消息序列，这些消息按顺序添加到一个叫做commit log的文件中。每个分区中的消息都有一个唯一的编号，称之为offset，用来唯一标示某个分区中的消息。每个分区，都对应一个commit log文件。一个分区中的消息的offset都是唯一的，但是不同的分区中的消息的offset可能是相同的。

日志

kafka一般不会删除消息，不管这些消息有没有被消费。只会根据配置的日志保留时间(log.retention.hours)确认消息多久被删除，默认保留最近一周的日志消息。kafka的性能与保留的消息数据量大小没有关系，因此保存大量的数据消息日志信息不会有什么影响。

每个消费者是基于自己在commit log中的消费进度(offset)来进行工作的。在kafka中，消费offset由消费者自己来维护；一般情况下我们按照顺序逐条消费commit log中的消息，当然我可以通过指定offset来重复消费某些消息，或者跳过某些消息。这意味kafka中的消费者对集群的影响是非常小的，添加一个或者减少一个消费者，对于集群或者其他消费者来说，都是没有影响的，因为每个消费者维护各自的消费offset。

九、Topic/Partition/Broker的初体验

创建多个分区的主题

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 2 --topic test1

查看下topic的情况

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test1

第一行是所有分区的概要信息，之后的每一行表示每一个partition的信息。

leader节点负责给定partition的所有读写请求。
replicas表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

创建的名称为"test"的topic

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test

之前设置了topic的partition数量为1，备份因子为1，因此显示就如上所示了。

查询kafka的数据文件存储目录

cat /opt/kafka_2.13-2.7.1/config/server.properties

找到log.dirs配置的路径

进入kafka的数据文件存储目录查看test和test1主题的消息日志文件

消息日志文件主要存放在分区文件夹里的以log结尾的日志文件里

增加topic的分区数量(目前kafka不支持减少分区)

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh -alter --partitions 3 --zookeeper 106.14.132.94:2181 --topic test

查看分区

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test

一个主题，代表逻辑上的一个业务数据集，比如按数据库里不同表的数据操作消息区分放入不同主题，订单相关操作消息放入订单主题，用户相关操作消息放入用户主题，对于大型网站来说，后端数据都是海量的，订单消息很可能是非常巨量的，比如有几百个G甚至达到TB级别，如果把这么多数据都放在一台机器上可定会有容量限制问题，那么就可以在主题内部划分多个分区来分片存储数据，不同的分区可以位于不同的机器上，每台机器上都运行一个Kafka的进程Broker。

为什么要对主题下数据进行分区存储？

commit log文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，相当于对数据做了分布式存储，理论上一个主题可以处理任意数量的数据。
为了提高并行度。

十、消费顺序

一个分区同一个时刻在一个消费组中只能有一个消费者实例在消费，从而保证消费顺序。消费组中的消费者实例的数量不能比一个主题中的分区的数量多，否则，多出来的消费者消费不到消息。Kafka只在分区的范围内保证消息消费的局部顺序性，不能在同一个主题中的多个分区中保证总的消费顺序性。如果有在总体上保证消费顺序的需求，那么我们可以通过将主题的分区数量设置为1，将消费组中的消费者实例数量也设置为1，但是这样会影响性能，所以kafka的顺序消费很少用。

十一、Kafka核心总控制器Controller

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（Kafka Controller），它负责管理整个集群中所有分区和副本的状态。

当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

十二、Controller选举机制

在kafka集群启动的时候，会自动选举一台broker作为总控制器来管理整个集群，选举的过程是集群中每个broker都会尝试在zookeeper上创建一个/controller临时节点，zookeeper会保证只有一个broker能创建成功，这个broker就会成为集群的总控制器。当这个总控制器角色的broker宕机了，此时zookeeper临时节点会消失，集群里其他broker会一直监听这个临时节点，发现临时节点消失了，就竞争再次创建临时节点，就是我们上面说的选举机制，zookeeper又会保证有一个broker成为新的controller。

具备控制器身份的broker需要比其他普通的broker多一份职责，具体细节如下：

监听broker相关的变化。为Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。
监听topic相关的变化。为Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；为Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。
从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对于所有topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。
更新集群的元数据信息，同步到其他普通的broker节点中。

十三、Partition副本选举Leader机制

控制器感知到分区leader所在的broker挂了(控制器监听了很多zk节点可以感知到broker存活)，控制器会从ISR列表(参数unclean.leader.election.enable=false的前提下)里挑第一个broker作为leader(第一个broker最先放进ISR列表，可能是同步数据最多的副本)，如果参数unclean.leader.election.enable为true，代表在ISR列表里所有副本都挂了的时候可以在ISR列表以外的副本中选leader，这种设置，可以提高可用性，但是选出的新leader有可能数据少很多。

副本进入ISR列表有两个条件：

副本节点不能产生分区，必须能与zookeeper保持会话以及跟leader副本网络连通
副本能复制leader上的所有写操作，并且不能落后太多。(与leader副本同步滞后的副本，是由replica.lag.time.max.ms 配置决定的，超过这个时间都没有跟leader同步过的一次的副本会被移出ISR列表)

十四、消费者消费消息的offset记录机制

每个消费者会定期将自己消费分区的offset提交给kafka内部topic： consumer_offsets，提交过去的时候，key是consumerGroupId+topic+分区号，value就是当前offset的值，kafka会定期清理topic里的消息，最后就保留最新的那条数据因为consumer_offsets可能会接收高并发的请求，kafka默认给其分配50个分区(可以通过offsets.topic.num.partitions设置)，这样可以通过加机器的方式抗大并发。通过如下公式可以选出consumer消费的offset要提交到 consumer_offsets的哪个分区，公式：hash(consumerGroupId) % consumer_offsets主题的分区数

十五、消费者Rebalance机制

rebalance就是说如果消费组里的消费者数量有变化或消费的分区数有变化，kafka会重新分配消费者消费分区的关系。比如consumer
group中某个消费者挂了，此时会自动把分配给他的分区交给其他的消费者，如果他又重启了，那么又会把一些分区重新交还给他。

注意：rebalance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebanlance。
如下情况可能会触发消费者rebalance：

消费组里的consumer增加或减少了。
动态给topic增加了分区
消费组订阅了更多的topic。

rebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS会有影响，如果kafka集群内节点较多，比如数百个，那重平衡可能会耗时极多，所以应尽量避免在系统高峰期的重平衡发生。

十六、消费者Rebalance分区分配策略

主要有三种rebalance的策略：range、round-robin、sticky。
Kafka 提供了消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。默认情况为range分配策略。
假设一个主题有10个分区(0-9)，现在有三个consumer消费：
range策略就是按照分区序号排序，假设 n＝分区数／消费者数量 = 3， m＝分区数%消费者数量 = 1，那么前 m 个消费者每个分配 n+1 个分区，后面的（消费者数量－m ）个消费者每个分配 n 个分区。
比如分区03给一个consumer，分区46给一个consumer，分区7~9给一个consumer。
round-robin策略就是轮询分配，比如分区0、3、6、9给一个consumer，分区1、4、7给一个consumer，分区2、5、8给一个consumer
sticky策略初始时分配策略与round-robin类似，但是在rebalance的时候，需要保证如下两个原则。
1）分区的分配要尽可能均匀。
2）分区的分配尽可能与上次分配的保持相同。
当两者发生冲突时，第一个目标优先于第二个目标。这样可以最大程度维持原来的分区分配的策略。
比如对于第一种range情况的分配，如果第三个consumer挂了，那么重新用sticky策略分配的结果如下：
consumer1除了原有的0~3，会再分配一个7
consumer2除了原有的4~6，会再分配8和9

十七、Rebalance过程

第一阶段：选择组协调器

组协调器GroupCoordinator

每个消费组都会选择一个broker作为自己的组协调器coordinator，负责监控这个消费组里的所有消费者的心跳，以及判断是否宕机，然后开启消费者rebalance。消费组中的每个消费者启动时会向kafka集群中的某个节点发送 FindCoordinatorRequest 请求来查找对应的组协调器GroupCoordinator，并跟其建立网络连接。

组协调器选择方式

消费者消费的offset要提交到__consumer_offsets的哪个分区，这个分区leader对应的broker就是这个消费组的coordinator

第二阶段：加入消费组JOIN GROUP

在成功找到消费组所对应的 GroupCoordinator 之后就进入加入消费组的阶段，在此阶段的消费者会向 GroupCoordinator 发送 JoinGroupRequest 请求，并处理响应。然后GroupCoordinator 从一个consumer group中选择第一个加入group的consumer作为leader(消费组协调器)，把consumer group情况发送给这个leader，接着这个leader会负责制定分区方案。

第三阶段（ SYNC GROUP)

consumer leader通过给GroupCoordinator发送SyncGroupRequest，接着GroupCoordinator就把分区方案下发给各个消费者，他们会根据指定分区的leader broker进行网络连接以及消息消费。

十八、producer发布消息机制

写入方式

producer 采用 push 模式将消息发布到 broker，每条消息都被 append 到 patition 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。

消息路由

producer 发送消息到 broker 时，会根据分区算法选择将其存储到哪一个 partition。其路由机制为：

指定了 patition，则直接使用；
未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition
patition 和 key 都未指定，使用轮询选出一个 patition。

写入流程

producer 先从 zookeeper 的 "/brokers/.../state" 节点找到该 partition 的 leader。
producer 将消息发送给该 leader。
leader 将消息写入本地 log。
followers 从 leader pull 消息，写入本地 log 后向leader 发送 ACK。
leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset）并向 producer 发送 ACK。

十九、HW与LEO

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。

下图详细的说明了当producer生产消息至broker后，ISR以及HW和LEO的流转过程

由此可见，Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式极大的影响了吞吐率。而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。再回顾下消息发送端对发出消息持久化机制参数acks的设置，我们结合HW和LEO来看下acks=1的情况。

结合HW和LEO看下 acks=1的情况

二十、日志分段存储

Kafka 一个分区的消息数据对应存储在一个文件夹下，以topic名称+分区号命名，消息在分区内是分段(segment)存储，每个段的消息都存储在不一样的log文件里，这种特性方便oldsegment file快速被删除，kafka规定了一个段位的 log 文件最大为 1G，做这个限制目的是为了方便把 log文件加载到内存去操作。

# 部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index文件，# 如果要定位消息的offset会先在这个文件里快速定位，再去log文件里找具体消息00000000000000000000.index# 消息存储文件，主要存offset和消息体00000000000000000000.log# 消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset到timeindex文件，# 如果需要按照时间来定位消息的offset，会先在这个文件里查找00000000000000000000.timeindex00000000000005367851.index00000000000005367851.log00000000000005367851.timeindex00000000000009936472.index00000000000009936472.log00000000000009936472.timeindex

这个 9936472 之类的数字，就是代表了这个日志段文件里包含的起始 Offset，也就说明这个分区里至少都写入了接近 1000万条数据了。Kafka Broker 有一个参数，log.segment.bytes，限定了每个日志段文件的大小，最大就是 1GB。一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做 log rolling，正在被写入的那个日志段文件，叫做 active log segment。

二十一、十亿消息数据线上环境规划

二十二、JVM参数设置

kafka是scala语言开发，运行在JVM上，需要对JVM参数合理设置，参看JVM调优专题
修改bin/kafka-start-server.sh中的jvm设置，假设机器是32G内存，可以如下设置：

export KAFKA_HEAP_OPTS="-Xmx16G -Xms16G -Xmn10G -XX:MetaspaceSize=256M -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:G1HeapRegionSize=16M"

这种大内存的情况一般都要用G1垃圾收集器，因为年轻代内存比较大，用G1可以设置GC最大停顿时间，不至于一次minor gc就花费太长时间，当然，因为像kafka，rocketmq，es这些中间件，写数据到磁盘会用到操作系统的page cache，所以JVM内存不宜分配过大，需要给操作系统的缓存留出几个G。

二十三、线上问题及优化

一、消息丢失情况：

一、消息发送端

（1）acks=0：表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。大数据统计报表场景，对性能要求很高，对数据丢失不敏感的情况可以用这种。
（2）acks=1：至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
（3）acks=-1或all：这意味着leader需要等待所有备份(min.insync.replicas配置的备份个数)都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。当然如果min.insync.replicas配置的是1则也可能丢消息，跟acks=1情况类似。

二、消息消费端

如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了，但是此时你consumer直接宕机了，未处理完的数据丢失了，下次也消费不到了。

二、消息重复消费

一、消息发送端

发送消息如果配置了重试机制，比如网络抖动时间过长导致发送端发送超时，实际broker可能已经接收到消息，但发送方会重新发送消息。

二、消息消费端

如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理。一般消费端都是要做消费幂等处理的。

三、消息乱序

如果发送端配置了重试机制，kafka不会等之前那条消息完全发送成功才去发送下一条消息，这样可能会出现，发送了1，2，3条消息，第一条超时了，后面两条发送成功，再重试发送第1条消息，这时消息在broker端的顺序就是2，3，1了
所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式去发消息，当然acks不能设置为0，这样也能保证消息发送的有序。
kafka保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消费发到内存队列(可以搞多个)，一个内存队列开启一个线程顺序处理消息。

四、消息积压

1）线上有时因为发送方发送消息速度过快，或者消费方处理消息过慢，可能会导致broker积压大量未消费消息。
此种情况如果积压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到的消息快速转发到其他topic(可以设置很多分区)，然后再启动多个消费者同时消费新主题的不同分区。

2）由于消息数据格式变动或消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未消费消息。
此种情况可以将这些消费不成功的消息转发到其它队列里去(类似死信队列)，后面再慢慢分析死信队列里的消息处理问题。

五、延时队列

延时队列存储的对象是延时消息。所谓的“延时消息”是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多，比如：
1）在订单系统中，一个用户下单之后通常有 30 分钟的时间进行支付，如果 30 分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了。
2）订单完成1小时后通知用户进行评价。

实现思路：发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中（topic_1s，topic_5s，topic_10s，...topic_2h，这个一般不能支持任意时间段的延时），然后通过定时器进行轮训消费这些topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了。

六、消息回溯

如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序bug导致的计算错误，当程序bug修复后，这时可能需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消费，这种可以用consumer的offsetsForTimes、seek等方法指定从某个offset偏移的消息开始消费，参见上面的内容。

二十四、分区数越多吞吐量越高吗？

可以用kafka压测工具自己测试分区数不同，各种情况下的吞吐量

# 往test里发送一百万消息，每条设置1KB# throughput 用来进行限流控制，当设定的值小于 0 时不限流，当设定的值大于 0 时，当发送的吞吐量大于该值时就会被阻塞一段时间bin/kafka-producer-perf-test.sh --topic test --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=106.14.132.94:9092 acks=1

网络上很多资料都说分区数越多吞吐量越高，但从压测结果来看，分区数到达某个值吞吐量反而开始下降，实际上很多事情都会有一个临界值，当超过这个临界值之后，很多原本符合既定逻辑的走向又会变得不同。一般情况分区数跟集群机器数量相当就差不多了。
当然吞吐量的数值和走势还会和磁盘、文件系统、 I/O调度策略等因素相关。
注意：如果分区数设置过大，比如设置10000，可能会设置不成功，后台会报错"java.io.IOException : Too many open files"。
异常中最关键的信息是“ Too many open flies”，这是一种常见的 Linux 系统错误，通常意味着文件描述符不足，它一般发生在创建线程、创建 Socket、打开文件这些场景下。在 Linux系统的默认设置下，这个文件描述符的个数不是很多，通过 ulimit -n 命令可以查看：一般默认是1024，可以将该值增大，比如：ulimit -n 65535

二十五、消息传递保障

at most once(消费者最多收到一次消息，0--1次)：acks = 0 可以实现。
at least once(消费者至少收到一次消息，1--多次)：ack = all 可以实现。
exactly once(消费者刚好收到一次消息)：at least once 加上消费者幂等性可以实现，还可以用kafka生产者的幂等性来实现。

kafka生产者的幂等性：因为发送端重试导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息只接收一次，只需在生产者加上参数 props.put(“enable.idempotence”, true) 即可，默认是false不开启。
具体实现原理是，kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker，broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查PID和Sequence Number，如果相同不会再接收。

PID：每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，这个PID对用户完全是透明的。生产者如果重启则会生成新的PID。 Sequence Number：对于每个 PID，该 Producer 发送到每个
Partition 的数据都有对应的序列号，这些序列号是从0开始单调递增的。

二十六、kafka的事务

Kafka的事务不同于Rocketmq，Rocketmq是保障本地事务(比如数据库)与mq消息发送的事务一致性，Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)，一般在kafka的流式计算场景用得多一点，比如，kafka需要对一个topic里的消息做不同的流式计算处理，处理完分别发到不同的topic里，这些topic分别被不同的下游系统消费(比如hbase，redis，es等)，这种我们肯定希望系统发送到多个topic的数据保持事务一致性。Kafka要实现类似Rocketmq的分布式事务需要额外开发功能。
kafka的事务处理可以参考官方文档

 Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("transactional.id", "my-transactional-id"); Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()); //初始化事务 producer.initTransactions(); try {     //开启事务     producer.beginTransaction();     for (int i = 0; i < 100; i++){         //发到不同的主题的不同分区         producer.send(new ProducerRecord<>("hdfs-topic", Integer.toString(i), Integer.toString(i)));         producer.send(new ProducerRecord<>("es-topic", Integer.toString(i), Integer.toString(i)));         producer.send(new ProducerRecord<>("redis-topic", Integer.toString(i), Integer.toString(i)));     }     //提交事务     producer.commitTransaction(); } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {     // We can't recover from these exceptions, so our only option is to close the producer and exit.     producer.close(); } catch (KafkaException e) {     // For all other exceptions, just abort the transaction and try again.     //回滚事务     producer.abortTransaction(); } producer.close();

二十七、kafka高性能的原因

磁盘顺序读写：kafka消息不能修改以及不会从文件中间删除保证了磁盘顺序读，kafka的消息写入文件都是追加在文件末尾，不会写入文件中的某个位置(随机写)保证了磁盘顺序写。
数据传输的零拷贝。
读写数据的批量batch处理以及压缩传输。

数据传输零拷贝原理：

二十八、Java客户端访问Kafka

kafka官方文档给的demo

以下的代码是从kafka的官方摘录过来，并且进行翻译的，可用于快速集成开发。

引入maven依赖

<dependency>   <groupId>org.apache.kafka</groupId>   <artifactId>kafka-clients</artifactId>   <version>2.4.1</version></dependency>

消息发送端代码

package com.sky.kafka.kafkaDemo;import com.alibaba.fastjson.JSON;import org.apache.kafka.clients.producer.*;import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;import java.util.concurrent.CountDownLatch;import java.util.concurrent.ExecutionException;import java.util.concurrent.TimeUnit;public class MsgProducer {    private final static String TOPIC_NAME = "my-replicated-topic";    public static void main(String[] args) throws InterruptedException, ExecutionException {        Properties props = new Properties();        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");         /*         发出消息持久化机制参数        （1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。        （2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一             条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。        （3）acks=-1或all： 需要等待 min.insync.replicas(默认为1，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略会保证            只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。         */        /*props.put(ProducerConfig.ACKS_CONFIG, "1");         *//*        发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在        接收者那边做好消息接收的幂等性处理        *//*        props.put(ProducerConfig.RETRIES_CONFIG, 3);        //重试间隔设置        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);        //设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);        *//*        kafka本地线程会从缓冲区取数据，批量发送到broker，        设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去        *//*        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);        *//*        默认值是0，意思就是消息必须立即被发送，但这样会影响性能        一般设置10毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去        如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长        *//*        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);*/        //把发送的key从字符串序列化为字节数组        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        //把发送消息value从字符串序列化为字节数组        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        Producer<String, String> producer = new KafkaProducer<String, String>(props);        int msgNum = 5;        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);        for (int i = 1; i <= msgNum; i++) {            Order order = new Order(i, 100 + i, 1, 1000.00);            //指定发送分区            /*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));*/            //未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME                    , order.getOrderId().toString(), JSON.toJSONString(order));            //等待消息发送成功的同步阻塞方法            /*RecordMetadata metadata = producer.send(producerRecord).get();            System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"                    + metadata.partition() + "|offset-" + metadata.offset());*/            //异步回调方式发送消息            producer.send(producerRecord, new Callback() {                public void onCompletion(RecordMetadata metadata, Exception exception) {                    if (exception != null) {                        System.err.println("发送消息失败：" + exception.getStackTrace());                    }                    if (metadata != null) {                        System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"                                + metadata.partition() + "|offset-" + metadata.offset());                    }                    countDownLatch.countDown();                }            });            //送积分 TODO        }        countDownLatch.await(5, TimeUnit.SECONDS);        producer.close();    }}

消息接收端代码

package com.sky.kafka.kafkaDemo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class MsgConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        // 是否自动提交offset，默认就是true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自动提交offset的间隔时间
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /*
        当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
        latest(默认) ：只消费自己启动之后发送到主题的消息
        earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      /*
      consumer给broker发送心跳的间隔时间，broker接收到心跳如果此时有rebalance发生会通过心跳响应将
      rebalance方案下发给consumer，这个时间可以稍微短一点
      */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
        服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其踢出消费组，
        对应的Partition也会被重新分配给其他consumer，默认是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        //一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        /*
        如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
        会将其踢出消费组，将分区分配给别的consumer消费
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消费指定分区
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //消息回溯消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

        //指定offset消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

        //从指定时间点开始消费
        /*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //从1小时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(topicName, par.partition()), fetchDataTime);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }*/

        while (true) {
            /*
             * poll() API 是拉取消息的长轮询
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到消息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            }

            /*if (records.count() > 0) {
                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
                consumer.commitSync();

                // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });

            }*/
        }
    }
}

腾讯云CKafka使用SDK收发消息

步骤一：添加 Java 依赖库

在 pom.xml 中添加以下依赖。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.2</version>
</dependency>

步骤二：准备配置

创建 JAAS 配置文件 ckafka_client_jaas.conf。

KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="yourinstance#yourusername"
password="yourpassword";
};

创建消息队列 CKafka 配置文件 kafka.properties。

## 配置接入网络，在控制台的实例详情页面接入方式模块的网络列复制。
bootstrap.servers=xx.xx.xx.xx:xxxx
## 配置Topic，在控制台上topic管理页面复制。
topic=XXX
## 配置Consumer Group，您可以自定义设置
group.id=XXX
##JAAS配置文件ckafka_client_jaas.conf的路径。
java.security.auth.login.config.plain=/xxxx/ckafka_client_jaas.conf

创建配置文件加载程序 CKafkaConfigurer.java

package com.sky.kafka.kafkaDemo;

import java.util.Properties;

/**
 * @author zhiwei Liao
 * @version 1.0
 * @Description
 * @Date 2021/12/16 17:19
 */

public class CKafkaConfigurer {
    private static Properties properties;
    public static void configureSaslPlain() {
        //如果用-D或者其它方式设置过，这里不再设置。
        if (null == System.getProperty("java.security.auth.login.config")) {
            //请注意将XXX修改为自己的路径。
            System.setProperty("java.security.auth.login.config",
                    getCKafkaProperties().getProperty("java.security.auth.login.config.plain"));
        }
    }
    public synchronized static Properties getCKafkaProperties() {
        if (null != properties) {
            return properties;
        }
        //获取配置文件kafka.properties的内容。
        Properties kafkaProperties = new Properties();
        try {
            kafkaProperties.load(KafkaSaslProducerDemo.class.getClassLoader().getResourceAsStream("kafka.properties"));
        } catch (Exception e) {
            System.out.println("getCKafkaProperties error");
        }
        properties = kafkaProperties;
        return kafkaProperties;
    }
}

步骤三：发送消息

创建发送消息程序 KafkaSaslProducerDemo.java。

package com.sky.kafka.kafkaDemo;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @author zhiwei Liao
 * @version 1.0
 * @Description
 * @Date 2021/12/16 17:19
 */

public class KafkaSaslProducerDemo {
    public static void main(String args[]) {
        //设置JAAS配置文件的路径。
        CKafkaConfigurer.configureSaslPlain();
        //加载kafka.properties。
        Properties kafkaProperties =  CKafkaConfigurer.getCKafkaProperties();
        Properties props = new Properties();
        //设置接入点，请通过控制台获取对应Topic的接入点。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        //接入协议。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        //Plain方式。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        //消息队列Kafka版消息的序列化方式。
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //请求的最长等待时间。
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        //设置客户端内部重试次数。
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        //设置客户端内部重试间隔。
        props.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 3000);
        //构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可。
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        //构造一个消息队列Kafka版消息。
        String topic = kafkaProperties.getProperty("topic"); //消息所属的Topic，请在控制台申请之后，填写在这里。
        String value = "this is ckafka msg value"; //消息的内容。
        try {
            //批量获取Future对象可以加快速度。但注意，批量不要太大。
            List<Future<RecordMetadata>> futures = new ArrayList<>(128);
            for (int i =0; i < 100; i++) {
                //发送消息，并获得一个Future对象。
                ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic, value + ": " + i);
                Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
                futures.add(metadataFuture);
            }
            producer.flush();
            for (Future<RecordMetadata> future: futures) {
                //同步获得Future对象的结果。
                RecordMetadata recordMetadata = future.get();
                System.out.println("Produce ok:" + recordMetadata.toString());
            }
        } catch (Exception e) {
            //客户端内部重试之后，仍然发送失败，业务要应对此类错误。
            System.out.println("error occurred");
        }
    }
}

编译并运行KafkaSaslProducerDemo.java发送消息。

运行结果（输出）。

Produce ok:ckafka-topic-demo-0@198Produce ok:ckafka-topic-demo-0@199

在 CKafka 控制台topic管理页面，选择对应的 topic，点击更多 > 消息查询，查看刚刚发送的消息。

步骤四：消费消息

创建 Consumer 订阅消息程序 KafkaSaslConsumerDemo.java。

package com.sky.kafka.kafkaDemo;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author zhiwei Liao
 * @version 1.0
 * @Description
 * @Date 2021/12/16 17:20
 */

public class KafkaSaslConsumerDemo {
    public static void main(String args[]) {
        //设置JAAS配置文件的路径。
        CKafkaConfigurer.configureSaslPlain();
        //加载kafka.properties。
        Properties kafkaProperties =  CKafkaConfigurer.getCKafkaProperties();
        Properties props = new Properties();
        //设置接入点，请通过控制台获取对应Topic的接入点。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        //接入协议。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        //Plain方式。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        //两次Poll之间的最大允许间隔。
        //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        //每次Poll的最大数量。
        //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        //消息的反序列化方式。
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        //当前消费实例所属的消费组，请在控制台申请之后填写。
        //属于同一个组的消费实例，会负载消费消息。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        //构造消费对象，也即生成一个消费实例。
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        //设置消费组订阅的Topic，可以订阅多个。
        //如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
        List<String> subscribedTopics =  new ArrayList<String>();
        //如果需要订阅多个Topic，则在这里添加进去即可。
        //每个Topic需要先在控制台进行创建。
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic: topics) {
            subscribedTopics.add(topic.trim());
        }
        consumer.subscribe(subscribedTopics);
        //循环消费消息。
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("消费了，Consume partition:%d offset:%d", record.partition(), record.offset()));
                }
            } catch (Exception e) {
                System.out.println("consumer error!");
            }
        }
    }
}

编译并运行 KafkaSaslConsumerDemo.java 消费消息。

运行结果。

Consume partition:0 offset:298
Consume partition:0 offset:299

在 CKafka 控制台Consumer Group页面，选择对应的消费组名称，在主题名称输入 topic 名称，点击查询详情，查看消费详情。

腾讯云CKafka文档

腾讯云CKafka接入方式

公网域名接入方式就是以上需要携带用户名和密码的方式进行的，VPC网络则是以内网的方式进行，不需要用户名和密码，一般环境部署都会使用VPC网络接入提高性能，加上权限校验会减低性能。
一般而已，我们可以通过kafka客户端连接工具测试kafka，这里提供下载地址：
链接： kafka下载
提取码：2021

客户端连接公网域名接入

填写：

org.apache.kafka.common.security.plain.PlainLoginModule required username="实例id#用户名" password="密码";

无用户名密码的，可以直接连接

二十九、Spring Boot整合Kafka

引入spring boot kafka依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml配置

server:
  port: 8080

spring:
  kafka:
    bootstrap-servers: 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094
    producer: # 生产者
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384
      buffer-memory: 33554432
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: default-group
      enable-auto-commit: false
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # RECORD
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
      # BATCH
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
      # TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
      # COUNT
      # TIME |　COUNT　有一个条件满足时提交
      # COUNT_TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # MANUAL
      # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
      # MANUAL_IMMEDIATE
      ack-mode: manual_immediate

发送者代码

package com.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class KafkaController {

    private final static String TOPIC_NAME = "my-replicated-topic";

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/send")
    public void send() {
        kafkaTemplate.send(TOPIC_NAME, 0, "key", "this is a msg");
    }

}

消费者代码

package com.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyConsumer {

    /**
     * @KafkaListener(groupId = "testGroup", topicPartitions = {
     *             @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     *             @TopicPartition(topic = "topic2", partitions = "0",
     *                     partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
     *     },concurrency = "6")
     *  //concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
     * @param record
     */
    @KafkaListener(topics = "my-replicated-topic",groupId = "zhugeGroup")
    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        //手动提交offset
        ack.acknowledge();
    }

    /*//配置多个消费组
    @KafkaListener(topics = "my-replicated-topic",groupId = "tulingGroup")
    public void listenTulingGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);
        ack.acknowledge();
    }*/
}

文章总结、点赞、收藏、留言、关注、聊聊我自己.png

点击进入我的【自媒体矩阵】