• kafka是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。

传统消息队列的应用场景

• 使用消息队列的好处：
  • 解耦：允许你独立地扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
  • 可恢复性：系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
  • 缓冲：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
  • 削峰：在访问量剧增的情况下，使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
  • 异步通信：很多时候用户不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。
• 消息队列的两种模式:
  • 点对点模式：一对一，消费者主动拉取消息，消费者消费消息后，消息就被清除。队列支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
  • 发布/订阅模式：一对多，生产者生产消息放到队列中，队列推送消息，消费者消费消息后，消息不会被清除。

点对点模式

发布/订阅模式（缺点：消费者长轮询）

kafka基础架构

• Producer：消息生产者，就是向 kafka broker 发消息的客户端；
• Consumer：消息消费者，向 kafka broker 取消息的客户端；
• Consumer Group（CG）：消费者组，由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内某个消费者消费；消费者组之间互不影响。 所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
• Broker：一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多个 topic。
• Topic： 可以理解为一个队列， 生产者和消费者面向的都是一个 topic；
• Partition： 为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；
• Replication：副本，为保证集群中的某个节点发生故障时，该节点上的分区数据不丢失，且Kafka仍然能够继续工作， Kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower。
• Leader： 每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 leader。
• Follower：每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。 leader 发生故障时，某个 Follower 会成为新的 leader。

依赖zookeeper集群搭建kafka集群

• zookeeper集群搭建教程
• kafka安装包下载页面
• 准备3台节点：192.168.211.147，192.168.211.141，192.168.211.156，并在各个节点的/opt目录下解压压缩包：

tar -zxvf kafka_2.12-2.5.0.tgz、mv kafka_2.12-2.5.0 kafka-1
tar -zxvf kafka_2.12-2.5.0.tgz、mv kafka_2.12-2.5.0 kafka-2
tar -zxvf kafka_2.12-2.5.0.tgz、mv kafka_2.12-2.5.0 kafka-3

• 修改各个节点中kafka服务器的配置文件server.properties、producer.properties、consumer.properties：
• /opt/kafka-1/config/server.properties：

broker.id=1 #服务器节点编号
listeners=PLAINTEXT://192.168.211.147:9092 #设置监听哪个端口
advertised.listeners=PLAINTEXT://192.168.211.147:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-1/data #存放kafka数据的目录
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.211.147:2181,192.168.211.147:2182,192.168.211.147:2183
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

• /opt/kafka-1/config/producer.properties：

bootstrap.servers=192.168.211.147:9092
compression.type=none

• /opt/kafka-1/config/consumer.properties：

bootstrap.servers=192.168.211.147:9092
group.id=test

• /opt/kafka-2/config/server.properties：

broker.id=2 #服务器节点编号
listeners=PLAINTEXT://192.168.211.141:9092 #设置监听哪个端口
advertised.listeners=PLAINTEXT://192.168.211.141:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-2/data #存放kafka数据的目录
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.211.147:2181,192.168.211.147:2182,192.168.211.147:2183
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

• /opt/kafka-2/config/producer.properties：

bootstrap.servers=192.168.211.141:9092
compression.type=none

• /opt/kafka-2/config/consumer.properties：

bootstrap.servers=192.168.211.141:9092
group.id=test

• /opt/kafka-3/config/server.properties：

broker.id=3 #服务器节点编号
listeners=PLAINTEXT://192.168.211.156:9092 #设置监听哪个端口
advertised.listeners=PLAINTEXT://192.168.211.156:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/kafka-3/data #存放kafka数据的目录
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
#log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.211.147:2181,192.168.211.147:2182,192.168.211.147:2183
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

• /opt/kafka-3/config/producer.properties：

bootstrap.servers=192.168.211.156:9092
compression.type=none

• /opt/kafka-3/config/consumer.properties：

bootstrap.servers=192.168.211.156:9092
group.id=test

• 分别在各个节点的kafka安装目录下创建两个文件夹：mkdir data logs
• 分别在各个节点的kafka安装目录下的data文件夹中创建一个文件meta.properties：
• /opt/kafka-1/data/meta.properties：

version=0
broker.id=1 #注意和server.properties中的broker.id的值相同

• /opt/kafka-2/data/meta.properties：

version=0
broker.id=2

• /opt/kafka-3/data/meta.properties：

version=0
broker.id=3

• 先启动zookeeper集群，再启动kafka集群（可以单独写一个shell脚本来启动和关闭集群），然后键入命令：jps查看kafka是否已启动。

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
nohup bin/kafka-server-start.sh config/server.properties 1>/dev/null 2>&1 &

• 进入kafka安装目录下的logs文件夹，通过文件server.log可以查看该kafka节点的运行日志。
• 创建主题（bigdata3），指定分区数2个和副本数1个（注意：分区数可以超过节点数，副本数不能超过节点数）：

bin/kafka-topics.sh --zookeeper 192.168.211.147:2181 --create --topic bigdata3 --partitions 2 --replication-factor 1

• 列举主题（存放在zk中）

bin/kafka-topics.sh --zookeeper 192.168.211.147:2181 --list

• 删除主题（bigdata3），删除后会被标记为删除状态，等cpu空闲时再自动清除

./bin/kafka-topics.sh --zookeeper 192.168.211.147:2181 --delete --topic bigdata3

• 主题分布详情：

./bin/kafka-topics.sh --zookeeper 192.168.211.147:2182 --describe --topic bigdata3

• 往bigdata3主题上发送消息（启动生产者）：

./bin/kafka-console-producer.sh --broker-list 192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092 --topic bigdata3

• 从bigdata3主题上读取消息（启动消费者）：

./bin/kafka-console-consumer.sh --bootstrap-server 192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092 --topic bigdata3 --from-beginning

kafka工作流程

• kafka中消息是以topic进行分类的，producer生产消息，consumer消费消息，都是面向topic的。（从命令行操作看出）
• topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该 log 文件中存储的就是producer 生产的数据。（topic = N partition，partition = log）

kafka文件存储机制

• 由于生产者生产的消息会不断追加到.log文件末尾， 为防止.log文件过大导致数据定位效率低下，kafka采取了分片和索引机制，将每个partition分为多个segment。每个segment对应两个文件——.index文件和.log文件。 这些文件位于一个文件夹下，该文件夹的命名规则为：topic name+分区序号。例如：first-topic这个topic有三个分区，则其对应的文件夹为first-topic-0，first-topic-1，first-topic-2。

00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log

.index文件和.log文件的结构示意图

• .index和.log文件以当前segment的第一条消息的offset命名。.index文件存储大量的索引信息，.log文件存储大量的数据，索引文件中的元数据指向对应数据文件中message的物理偏移地址。
• 主题分区的原因：
  • 方便在集群中扩展，每个partition可以通过调整以适应它所在的机器，而一个topic又可以有多个partition组成，因此整个集群就可以适应适合的数据了；
  • 可以提高并发，因为可以以partition为单位进行读写了。
• 分区的原则：
  • 指明partition的情况下，直接将指明的值作为partiton值；
  • 没有指明partition值但有key的情况下，将key的 hash 值与 topic 的 partition 总数进行取余得到的partition值；
  • 既没有partition值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition值，即round-robin算法。

生产者发送数据分装为ProducerRecord对象的API

• 生产者数据可靠性保证：为保证producer发送的数据，能可靠地发送到指定的 topic， topic 的每个partition收到producer 发送的数据后，都需要向 producer 发送ack（acknowledgement：确认收到），若producer收到 ack，则会进行下一轮的发送，否则重新发送数据。

生产者数据可靠性保证机制

• 何时发送ack？确保有follower与leader同步完成，leader再发送ack，这样才能保证leader挂掉之后，能在follower中选举出新的leader。
• 多少个follower同步完成之后发送ack？①半数以上的follower同步完成，即可发送ack；②全部的follower同步完成，才可以发送ack。

• kafka选取了第二种方案，原因：①为了容忍 n 台节点的故障，第一种方案需要 2n+1 个副本，而第二种方案只需要 n+1 个副本，又 kafka 的每个分区都有大量的数据， 因此，第一种方案会造成大量数据的冗余。②虽然第二种方案的网络延迟会比较高，但网络延迟对 kafka 的影响较小。
• kafka采用第二种方案之后，设想以下情景： leader 收到数据，所有 follower 都开始同步数据，但有一个 follower，因为某种故障迟迟不能与 leader 进行同步，那 leader 就要一直等下去，直到它完成同步，才能发送 ack。这个问题怎么解决呢？
  • leader 维护了一个动态的in-sync replica set（ISR），意为和 leader 保持同步的 follower 集合。当 ISR 中的 follower 完成数据的同步之后，就会给 leader 发送 ack。若follower长时间未向leader同步数据，则该follower将被踢出ISR，该时间阈值由replica.lag.time.max.ms参数设定。当leader 发生故障之后，就会从 ISR 中选举新的 leader。
  • replica.lag.time.max.ms：If a follower hasn't sent any fetch requests or hasn't consumed up to the leaders log end offset for at least this time, the leader will remove the follower from isr。TYPE: long。DEFAULT: 10000。
• 若对数据的可靠性要求不是很高，能够容忍少量数据的丢失，则没必要等 ISR 中的 follower 全部接收成功，因此，kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡，acks可以选择以下几个值：
  • 0：producer不等待 broker 的 ack，这一操作提供了一个最低的延迟，broker一接收到还没有写入磁盘就已经返回，当 broker 故障时有可能丢失数据；
  • 1：producer等待 broker 的 ack，partition的 leader 落盘成功后返回才ack，若在 follower同步成功之前 leader 故障，则将会丢失数据；
  • -1（all）：producer等待 broker 的 ack，partition的 leader 和 ISR 的follower 全部落盘成功后才返回 ack。但若在 follower 同步完成后，broker 发送 ack 之前， leader 发生故障，则会造成数据重复。

acks=1

acks=-1，可能造成数据重复写入

• 助记：返ACK前，0无落盘，1一落盘，-1全落盘，（落盘：消息存到本地）
• kafka数据一致性

• LEO（Log End Offset）：每个副本的最后一个offset。
• HW（High Watermark）：高水位，指消费者能见到的最大的offset，ISR 队列中最小的 LEO。
• follower故障：follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后， follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步；等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。
• leader故障：leader 发生故障之后，会从 ISR 中选出一个新的 leader，之后为保证多个副本之间的数据一致性， 其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader同步数据。
• 注意：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。
• 当服务器的 ACK 级别设置为-1（all），可以保证 Producer 到 Server 之间不会丢失数据，即At Least Once语义。当服务器的 ACK 级别设置为 0，可以保证生产者每条消息只会被发送一次，即At Most Once语义。
• At Least Once可以保证数据不丢失，但是不能保证数据不重复。At Most Once可以保证数据不重复，但是不能保证数据不丢失。 对于一些非常重要的信息，比如说交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once语义。
  • 在 0.11 版本以前的 Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。对于多个下游应用的情况，每个都需要单独做全局去重，这就对性能造成了很大影响。
  • 0.11 版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据， Server 端都只会持久化一条。幂等性结合At Least Once语义，就构成了 Kafka 的Exactly Once语义。即：At Least Once + 幂等性 = Exactly Once。
• 要启用幂等性，只需要将 Producer 的参数中enable.idempotence设置为 true 即可。 Kafka的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在初始化的时候会被分配一个 PID，发往同一 Partition 的消息会附带 Sequence Number。而Broker 端会对<PID, Partition, SeqNumber>做缓存，当具有相同主键的消息提交时， Broker 只会持久化一条（单分区单会话）。但是生产者重启时PID 就会变化，同时不同的 Partition 也具有不同主键，所以幂等性无法保证跨分区跨会话的 Exactly Once。
• 消费者消费方式：consumer 采用 pull（拉） 模式从 broker 中读取数据。
  • push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快的速度传递消息，但是这样很容易造成 consumer 来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。
  • pull （拉）模式不足之处是，若 kafka 没有数据，则消费者可能会陷入循环中，一直返回空数据。 针对这一点， Kafka 的消费者在消费数据时会传入一个时长参数 timeout，若当前没有数据可供消费，则consumer 会等待一段时间之后再返回，这段时长即为 timeout。
• 消费者分区分配策略：一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以会涉及到 partition 的分配问题，即确定哪个 partition 由哪个 consumer 来消费。Kafka 有两种分配策略：①round-robin循环；②range。
• 关于Roudn Robin重分配策略，其主要采用的是一种轮询的方式来分配所有的分区，主要实现的步骤如下：假设有三个topic：t0、t1和t2，这三个topic拥有的分区数分别为1、2和3，则总共有六个分区，这六个分区分别为：t0-0、t1-0、t1-1、t2-0、t2-1和t2-2。又假设我们有三个consumer：C0、C1和C2，它们订阅的情况为：C0订阅t0，C1订阅t0和t1，C2订阅t0、t1和t2，则这些分区的分配步骤如下：
  • 首先将所有的partition和consumer按照字典序进行排序，所谓的字典序，就是按照其名称的字符串顺序，那么上面的六个分区和三个consumer排序之后分别为：t0-0、t1-0、t1-1、t2-0、t2-1、t2-2和C0、C1、C2。
  • 然后按顺序轮询的方式将这六个分区分配给三个consumer，若当前consumer没有订阅当前分区所在的topic，则轮询地判断下一个consumer：
    • 尝试将t0-0分配给C0，由于C0订阅了t0，因而可以分配成功；
    • 尝试将t1-0分配给C1，由于C1订阅了t1，因而可以分配成功；
    • 尝试将t1-1分配给C2，由于C2订阅了t1，因而可以分配成功；
    • 尝试将t2-0分配给C0，由于C0没有订阅t2，因而会轮询下一个consumer；
    • 尝试将t2-0分配给C1，由于C1没有订阅t2，因而会轮询下一个consumer；
    • 尝试将t2-0分配给C2，由于C2订阅了t2，因而可以分配成功；
    • 同理由于t2-1和t2-2所在的topic都没有被C0和C1订阅，因而都不会分配成功，最终都会分配给C2。
    • 从上面的步骤分析可以看出，轮询的策略就是简单地将所有的partition和consumer按照字典序进行排序之后，然后依次将partition分配给各个consumer，若当前的consumer没有订阅当前的partition，则会轮询下一个consumer，直至最终将所有的分区都分配完毕。但是从上面的分配结果可以看出，轮询的方式会导致每个consumer所承载的分区数量不一致，从而导致各个consumer压力不均一。
    • 按照上述的步骤将所有的分区都分配完毕之后，最终分区的订阅情况如下：

• Range重分配策略：首先会计算各个consumer将会承载的分区数量，然后将指定数量的分区分配给该consumer。假设有两个consumer：C0和C1，两个topic：t0和t1，这两个topic分别都有三个分区，总共的分区有六个：t0-0、t0-1、t0-2、t1-0、t1-1和t1-2，则这些分区的分配步骤如下：
  • 注意：Range策略是按照topic进行分区分配的。拿主题t0来分析，其首先会获取t0的所有分区：t0-0、t0-1和t0-2，以及所有订阅了该topic的consumer：C0和C1，并且会将这些分区和consumer按照字典序进行排序；
  • 然后按照平均分配的方式计算每个consumer会得到多少个分区，若没有除尽，则会将多出来的分区依次计算到前面几个consumer。假设有3个分区和2个consumer，则每个consumer至少会得到1个分区，因为3除以2后还余1，所以会将剩余的1个分区分配给第一个consumer，最终C0从第0个分区开始分配2个分区，而C1从第2个分区开始分配1个分区；
  • 同理，按照上面的步骤依次进行后面topic的分配，最终上面六个分区的分配情况如下所示：

• 由于 consumer 在消费过程中可能会出现断电宕机等故障， consumer 恢复后，需要从故障前的位置继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。

• Kafka 0.9 版本之前， consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为__consumer_offsets。
• 修改配置文件 consumer.properties：exclude.internal.topics=false。
• 0.11.0.0 之前版本：

bin/kafka-console-consumer.sh --topic __consumer_offsets --zookeeper hadoop102:2181 --formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning

• 0.11.0.0 及之后版本：

bin/kafka-console-consumer.sh --topic __consumer_offsets --zookeeper hadoop102:2181 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning

• 消费者组案例：测试同一个消费者组中的消费者， 同一时刻只能有一个消费者消费一个消息。
  • 修改consumer.properties文件中的 group.id 属性：group.id=zhangsan
  • 创建主题：bigdata3
  • 启动生产者
  • 分别启动2个消费者
  • 在生产者窗口输入消息，观察两个消费者窗口，会发现两个消费者窗口中，只有一个才会弹出消息。

bin/kafka-topics.sh --zookeeper 192.168.211.147:2181 --create --topic bigdata3 --partitions 2 --replication-factor 1
--------------------------------------------------------
./bin/kafka-console-producer.sh --broker-list 192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092 --topic bigdata3
-------------------------------------------------------
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092 --topic bigdata3 --consumer.config ./config/consumer.properties

• kafka高效读写数据：
  • 顺序写磁盘：Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。 官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
  • 零复制技术：NIC network interface controller 网络接口控制器。

• zookeeper 在 kafka 中的作用：Kafka 集群中有一个 broker 会被选举为controller，负责管理集群中 broker 的上下线，所有 topic 的分区副本分配和 leader 选举等工作。controller 的管理工作都是依赖于 zookeeper 的。以下为 partition 的 leader 选举过程：

• Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在 Exactly Once 语义的基础上，生产和消费可以跨分区和跨会话，要么全部成功，要么全部失败。
• Producer 事务：
  • 为了实现跨分区跨会话的事务，需要引入一个全局唯一的 Transaction ID，并将 Producer 获得的PID 和Transaction ID 绑定。这样当Producer 重启后就可以通过正在进行的 TransactionID 获得原来的 PID。
  • 为了管理 Transaction， Kafka 引入了一个新的组件 Transaction Coordinator。 Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。 Transaction Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。
• Consumer 事务：无法保证 Commit 的信息被精确消费一次，这是由于 Consumer 可以通过 offset 访问任意信息，而且不同的 Segment File 生命周期不同，同一事务的消息可能会出现重启后被删除的情况。
• 消息发送流程：Kafka 的 Producer 发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和 Sender线程，以及一个线程共享变量——RecordAccumulator。 main 线程将消息发送给 RecordAccumulator， Sender 线程不断从 RecordAccumulator 中拉取消息发送到 kafka broker。

• 相关参数：①batch.size： 只有数据积累到 batch.size 之后， sender 才会发送数据。②linger.ms：如果数据迟迟未达到 batch.size， sender 等待 linger.time 之后就会发送数据。
• kafka应用实例：测试异步发送消息以及带回调函数的生产者。
• 导入依赖：

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

• MyProducer.java

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class MyProducer {
    /**
     * KafkaProducer：需要创建一个生产者对象，用来发送数据
     * ProducerConfig：获取所需的一系列配置参数
     * ProducerRecord：每条数据都要封装成一个 ProducerRecord 对象
     */
    public static void main(String[] args) {
        //1、创建kafka生产者的配置信息
        Properties props = new Properties();
        //2、指定连接的kafka集群 broker-list
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        //3、ACK应答级别
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        //4、重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        //5、批次大小：16k
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        //6、等待时间：1ms
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        //7、RecordAccumulator 缓冲区大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        //8、指定序列化的类
        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");
        //9、创建生产者对象
        Producer<String, String> producer = new KafkaProducer<>(props);
        //10、发送数据
        /*for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("bigdata3", "test-" + i,
                    "test-" + i));
        }*/
        for (int i = 0; i < 10; i++) {
            /**
             * 回调函数会在 producer 收到 ack 时调用，且为异步调用，
             * 该方法有两个参数，分别是 RecordMetadata 和 Exception，
             * 若 Exception 为 null，则说明消息发送成功；否则说明消息发送失败。
             * 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
             */
            producer.send(new ProducerRecord<>("bigdata3",
                    "test-" + i), (metadata, exception) -> {
                if (exception == null) {
                    System.out.println(metadata.partition() + " - " + metadata.offset());
                } else {
                    exception.printStackTrace();
                }
            });
        }
        //11、关闭资源
        producer.close();
    }
}

• kafka应用实例：测试自定义分区器的生产者。
• MyPartitioner.java

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
public class MyPartitioner implements Partitioner {
    /**
     * 具体内容填写可参考默认分区器：org.apache.kafka.clients.producer.internals.DefaultPartitioner
     * 然后Producer配置中注册使用：
     * props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class);
     */
    @Override
    public void configure(Map<String, ?> configs) {

    }
    //自定义存储在哪个分区
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        return 0;
    }
    @Override
    public void close() {
    }
}

• kafka应用实例：测试同步发送消息的生产者。
• 同步发送的意思就是一条消息发送之后，会阻塞当前线程，直至其返回 ack。由于 send 方法返回的是一个 Future 对象，根据 Futrue 对象的特点，只需再调用 Future 对象的 get 方法即可。
• SyncProducer.java

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class SyncProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        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");
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("bigdata3",
                    "test-" + i), (metadata, exception) -> {
                if (exception == null) {
                    System.out.println(metadata.partition() + " - " + metadata.offset());
                } else {
                    exception.printStackTrace();
                }
            }).get(); //调用future的get方法来实现同步发送消息
        }
        producer.close();
    }
}

• kafka应用实例：测试消费者消费消息。
• MyConsumer.java

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 java.util.Collections;
import java.util.Properties;
public class MyConsumer {
    /**
     * KafkaConsumer：需要创建一个消费者对象，用来消费数据
     * ConsumerConfig：获取所需的一系列配置参数
     * ConsuemrRecord：每条数据都要封装成一个 ConsumerRecord 对象
     */
    public static void main(String[] args) {
        //创建消费者的配置信息
        Properties props = new Properties();
        //连接的kafka集群，--bootstrap-server
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        //设置消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "zhangsan");
        //开启自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        //自动提交offset的时间间隔
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //重置消费者的offset，注意需要换组名才能出现 --from-beginning 的效果
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//默认值是 latest
        //key，value的反序列化类
        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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //订阅主题
        consumer.subscribe(Collections.singletonList("bigdata3"));
        while (true) {
            //死循环获取所有记录
            ConsumerRecords<String, String> records = consumer.poll(100); //100ms
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d，key = %s，value = %s\n", record.offset(), record.key(), record.value());
            }
        }
        //关闭连接
        //consumer.close();
    }
}

• 虽然自动提交 offset 十分便利，但由于其是基于时间提交的，开发人员难以把握offset 提交的时机，因此 Kafka 还提供了手动提交 offset 的 API。手动提交 offset 的方法有两种：①commitSync（同步提交）；②commitAsync（异步提交）
  • 相同点：都会将本次 poll 的一批数据最高的偏移量提交；
  • 不同点：commitSync 阻塞当前线程，一直到提交成功，并且失败会自动重试（由不可控因素导致，也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。
• SyncCommitOffset.java

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 java.util.Collections;
import java.util.Properties;
public class SyncCommitOffset {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "zhangsan");
        //关闭自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//默认值是 latest
        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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("bigdata3"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100); //100ms
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d，key = %s，value = %s\n", record.offset(), record.key(), record.value());
            }
            //同步提交，当前线程会阻塞直到 offset 提交成功
            consumer.commitSync();
        }
        //关闭连接
        //consumer.close();
    }
}

• 虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程直到提交成功，吞吐量会收到很大的影响，因此更多情况下，会选用异步提交 offset 的方式。
• AsyncCommitOffset.java

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
public class AsyncCommitOffset {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "zhangsan");
        //关闭自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//默认值是 latest
        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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("bigdata3"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100); //100ms
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d，key = %s，value = %s\n", record.offset(), record.key(), record.value());
            }
            // 异步提交
            consumer.commitAsync((offsets, exception) -> {
                if (exception != null) {
                    System.err.println("Commit failed for" + offsets);
                }
            });
        }
        //关闭连接
        //consumer.close();
    }
}

• 无论是同步提交还是异步提交 offset，都有可能会造成数据的漏消费或者重复消费。先提交 offset 后消费，有可能造成数据的漏消费；而先消费后提交 offset，有可能会造成数据的重复消费。
• kafka 0.9 版本之前，offset 存储在 zookeeper中；0.9 版本及之后，默认将 offset 存储在 kafka的一个内置的 topic 中。除此之外， Kafka 还可以选择自定义存储 offset（需要借助 ConsumerRebalanceListener接口）。offset 的维护相当繁琐， 因为需要考虑到消费者的 Rebalance。（当有新的消费者加入消费者组、已有的消费者推出消费者组或者所订阅的主题的分区发生变化，就会触发到分区的重新分配，重新分配的过程叫做 Rebalance。）
• 消费者发生 Rebalance 之后，每个消费者消费的分区就会发生变化。因此消费者要首先获取到自己被重新分配到的分区，并且定位到每个分区最近提交的 offset 位置继续消费。
• CustomSaveOffset.java

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
public class CustomSaveOffset {
    private static Map<TopicPartition, Long> currentOffset = new HashMap<>();
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "zhangsan");
        //关闭自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");//默认值是 latest
        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");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("bigdata3"),
                new ConsumerRebalanceListener() {
                    // 该方法会在 Rebalance 之前调用
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                        commitOffset(currentOffset);
                    }

                    // 该方法会在 Rebalance 之后调用
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                        currentOffset.clear();
                        for (TopicPartition partition : partitions) {
                            //定位到最近提交的 offset 位置继续消费
                            consumer.seek(partition, getOffset(partition));
                        }
                    }
                });
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100); //100ms
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d，key = %s，value = %s\n", record.offset(), record.key(), record.value());
            }
            //异步提交
            commitOffset(currentOffset);
        }
        //关闭连接
        //consumer.close();
    }
    // 获取某分区的最新 offset
    private static long getOffset(TopicPartition partition) {
        return 0;
    }
    // 提交该消费者所有分区的 offset（可将offset存入MySQL数据库中）
    private static void commitOffset(Map<TopicPartition, Long> currentOffset) {
    }
}

• 拦截器原理：Producer 拦截器（interceptor）是在 kafka 0.10 版本被引入的，主要用于实现 clients 端的定制化控制逻辑。对于 producer 而言， interceptor 使得用户在消息发送前以及 producer 回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer 允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链（interceptor chain）。Intercetpor 的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：
  • configure(configs)：获取配置信息和初始化数据时调用。
  • onSend(ProducerRecord)：该方法封装进 KafkaProducer.send 方法中，即它运行在用户主线程中。 Producer 确保在消息被序列化以及计算分区前调用该方法。 用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区， 否则会影响目标分区的计算。
  • onAcknowledgement(RecordMetadata, Exception)：该方法会在消息从 RecordAccumulator 成功发送到 Kafka Broker 之后，或者在发送过程中失败时调用，并且通常在 producer 回调逻辑触发之前。 onAcknowledgement 运行在producer 的 IO 线程中，因此不要在该方法中放入很多的逻辑，否则会拖慢 producer 的消息发送效率。
  • close()：关闭 interceptor，主要用于执行一些资源清理工作。
• 如上所述：interceptor 可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外，倘若指定了多个 interceptor，则 producer 将按照指定顺序调用它们，并仅仅是捕获每个 interceptor 可能抛出的异常记录到错误日志中而非在向上传递。
• kafka应用实例：实现一个简单的双 interceptor 组成的拦截链。
• 需求：第一个 interceptor 会在消息发送前将时间戳信息加到消息 value 的最前部；第二个 interceptor 会在消息发送后更新成功发送消息数或失败发送消息数。
• TimeInterceptor.java

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
//时间戳拦截器
public class TimeInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public void configure(Map<String, ?> configs) {
    }
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        //创建一个新的 record，把时间戳写入消息体的最前部
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(),
                "TimeInterceptor: " + System.currentTimeMillis() + "," + record.value());
    }
    @Override
    public void close() {
    }
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }
}

• CounterInterceptor.java

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;
//统计发送消息成功和发送失败消息数，并在 producer 关闭时打印这两个计数器
public class CounterInterceptor implements ProducerInterceptor<String, String> {
    private int errorCounter = 0;
    private int successCounter = 0;
    @Override
    public void configure(Map<String, ?> configs) {
    }
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        return record;
    }
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        //统计成功和失败的次数
        if (exception == null) {
            successCounter++;
        } else {
            errorCounter++;
        }
    }
    @Override
    public void close() {
        //保存结果
        System.out.println("Successful sent: " + successCounter);
        System.out.println("Failed sent: " + errorCounter);
    }
}

• InterceptorProducer.java

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
public class InterceptorProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.147:9092,192.168.211.141:9092,192.168.211.156:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        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");
        //构建拦截链
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.zzw.interceptor.TimeInterceptor");
        interceptors.add("com.zzw.interceptor.CounterInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("bigdata3", "zhangsan", "message-" + i);
            producer.send(record);
        }
        //一定要关闭 producer，这样才会调用 interceptor 的 close 方法
        producer.close();
    }
}

• Kafka Eagle是开源可视化和管理软件。它允许您查询、可视化、提醒和探索您的指标，无论它们存储在哪里。简单地说，它为您提供了将kafka集群数据转换为漂亮的图形和可视化的工具。
• Kafka Eagle官方主页、 Kafka Eagle下载页面、Kafka Eagle官方文档
• 安装并配置Kafka Eagle的环境变量：

tar -zxvf kafka-eagle-bin-2.0.3.tar.gz
cd kafka-eagle-bin-2.0.3/
tar -zxvf kafka-eagle-web-2.0.3-bin.tar.gz -C /opt/eagle
-----------------------------------------------------------------
sudo vim /etc/profile
-----------------------------------------------------------------
export JAVA_HOME=/opt/jdk1.8
export KE_HOME=/opt/eagle
export PATH=$PATH:$KE_HOME/bin:$JAVA_HOME/bin
-----------------------------------------------------------------
source /etc/profile

• 进入kafka-eagle的conf目录下修改配置文件system-config.properties：

######################################
# multi zookeeper & kafka cluster list （zookeeper地址）
######################################
kafka.eagle.zk.cluster.alias=cluster1
cluster1.zk.list=192.168.211.147:2181,192.168.211.147:2182,192.168.211.147:2183

######################################
# zookeeper enable acl
######################################
cluster1.zk.acl.enable=false
cluster1.zk.acl.schema=digest
cluster1.zk.acl.username=test
cluster1.zk.acl.password=test123

######################################
# broker size online list
######################################
cluster1.kafka.eagle.broker.size=20

######################################
# zk client thread limit  （zk线程数）
######################################
kafka.zk.limit.size=25

######################################
# kafka eagle webui port  （kafka eagle的端口）
######################################
kafka.eagle.webui.port=8048

######################################
# kafka jmx acl and ssl authenticate
######################################
cluster1.kafka.eagle.jmx.acl=false
cluster1.kafka.eagle.jmx.user=keadmin
cluster1.kafka.eagle.jmx.password=keadmin123
cluster1.kafka.eagle.jmx.ssl=false
cluster1.kafka.eagle.jmx.truststore.location=/Users/dengjie/workspace/ssl/certificates/kafka.truststore
cluster1.kafka.eagle.jmx.truststore.password=ke123456

######################################
# kafka offset storage （kafka offset保存的位置）
######################################
cluster1.kafka.eagle.offset.storage=kafka

######################################
# kafka metrics, 15 days by default （开启图表）
######################################
kafka.eagle.metrics.charts=true
kafka.eagle.metrics.retain=15

######################################
# kafka sql topic records max
######################################
kafka.eagle.sql.topic.records.max=5000

######################################
# delete kafka topic token
######################################
kafka.eagle.topic.token=keadmin

######################################
# kafka sasl authenticate
######################################
cluster1.kafka.eagle.sasl.enable=false
cluster1.kafka.eagle.sasl.protocol=SASL_PLAINTEXT
cluster1.kafka.eagle.sasl.mechanism=SCRAM-SHA-256
cluster1.kafka.eagle.sasl.jaas.co![20210109181652.png](https://upload-images.jianshu.io/upload_images/7810329-5a7c09b3311a2647.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
nfig=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-eagle";
cluster1.kafka.eagle.sasl.client.id=
cluster1.kafka.eagle.blacklist.topics=
cluster1.kafka.eagle.sasl.cgroup.enable=false
cluster1.kafka.eagle.sasl.cgroup.topics=

######################################
# kafka mysql jdbc driver address （不用预先创建数据库）
######################################
kafka.eagle.driver=com.mysql.jdbc.Driver
kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password=123456

• 打开Kafka broker JMX：Kafka Eagle获取监控数据是通过JMX（Java Managent ExtenSion）来实现的，所以需要打开Kafka broker 的 JMX，可以直接修改kafka-server-start.sh文件：vi $KAFKA_HOME/bin/kafka-server-start.sh：

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
fi
---------------------------------------------------------------------------
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
    export JMX_PORT="9999"
fi

• 接下来，按顺序进行以下操作：
  • 运行zookeeper集群；
  • 运行kafka集群之前，需要设置JMX_PORT，否则Kafka Eagle 后台提示连接失败；
  • 启动Kafka Eagle服务器；

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
nohup bin/kafka-server-start.sh config/server.properties 1>/dev/null 2>&1 &
./ke.sh start

启动成功

登录页面