Kafka-Eagle框架用于监控Kafka集群运行状况。官网https://kafka-eagle.org
前置mysql。
关闭集群 kf.sn stop 修改vim /bin/kafka-server-start.sh 并同步到其他节点 if [ "x$KAFKA_HEAP_OPTS " = "x" ] ; then
export KAFKA_HEAP_OPTS = "-server -Xms2G -Xmx2G XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 XX:IniyiatingHeapOccupancyPercent=70"
export JMX_PORT = "9999"
管网下载：https://kafka-eagle.org压缩包kafka-eagle-bin-2.0.8.tar.gz并解压。
efak.zk.cluster.alias = cluster1
cluster.zk.list = node1:2181,node2:2181,node3:2181/kafka
.. . .. .
cluster1.efak.offset.storage = kafka
.. . .. .
efak.driver = com.mysql.jdbc.Driver
efak.url = jdbc:mysql://localhost:3306/kafka-eagle/?useUnicode= true
efak.username = root
efak.password = root
添加环境变量：sudo vim /etc/profile.d/my_env.sh
export KE_HOME = /export/server/efak
export PATH = $PATH : $KE_HOME /bin
source /etc/profile 生效
浏览器输入ip:8048
kafka2.8.0之前由zookeeper存储元数据，运行时动态选举controller，由controller进行kafka集群管理。
kafka不在依赖外部框架，可独立运行 controller管理集群时，不在从zookeeper中读取数据，集群性能提高。 由于不依赖zookeeper，集群扩展不再受zookeeper读写能力限制 controller不再动态选举，而是由配置文件规定。不像之前随机controller节点高负载 解压kafka_2.12-3.0.0.tgz -C /export/server/ 设置kafka软连接 在node1上修改/export/server/kafka/config/karaft/server.properties并同步到集群中其他节点: 分发kafka配置，对node.id，advertised.listeners进行修改 初始化居群数据目录 首先生成存储目录唯一ID：bin/kafka-storage.sh random-uuid （eg：123456） 用该ID格式化kafka存储目录（其余节点）： 启动kafka集群node1，2，3 停止kafka集群 在home/xuyu/bin目录下创建文件kafka.sh：vim kafka.sh #! /bin/bash
case $1 in
"start") {
for i in node1 node2 node3
do
echo "--- start $i kafka ---"
ssh $i "/export/server/kafka/bin/kafka-server-start.sh -daemon /export/serverkafka/config/kraft/server.properties"
done
};;
"stop"){
for i in node1 node2 node3
do
echo "--- stop $i kafka ---"
ssh $i "/export/server/kafka/bin/kafka-server-stop.sh"
done
};;
esac
添加执行权限 启停集群命令 100万日活，每人每天100条日志，每天总共的日志条数是100万100条=1亿条。 1k=1m/s。
服务器台数 = 2 * （生产者峰值生成速率 * 副本数 / 100） + 1 = 2 *（20m/s * 2 / 100） + 1 = 3
kafka 按照顺序读写：机械硬盘和固态硬盘顺序读写速度差不多。随机读写：选固态硬盘。
kafka内存=堆内存(kafka内部配置)+页缓存(服务器内存)。
kafka堆内存建议每个节点：10-15G vim kafka-server-start.sh：
if [ "x$KAFKA_HEAP_OPTS " = "x" ] ; then
export KAFKA_HEAP_OPTS = "-Xmx10G -Xms10G"
fi
查看kakfa进程号：jps1G 25%）/节点数 = （101G 25%）/3=1G。建议服务器内存大于11G。
num.io.threads=8：负责写磁盘的线程数，整个参数值要占总核数的50%
网络带宽=峰值吞吐量=20MB/s。千兆网卡即可。
Updating Broker Config
默认32m 默认16k ，增加会导致延迟过高。
先开启幂等，kafka事务API：
单分区内有序；多分区，分区间无序(实现需要在接收端按顺序排序，效率低，见前文)
buffer.memory：发送消息缓冲区，默认32m batch.size：发送消息大小，默认16K，太卡会卡 linger.ms：发送时间间隔，默认0，一般5-100ms compression.type：压缩方式，默认none，会增加cpu开销 调整fetch.max.bytes大小，默认50m 调整max.poll.records大小，默认500条 生产者角度 acks设置为-1 幂等性(enable.idempotence=true)+事务 broker服务端角度 分区副本大于等于2(–replication-factor 2) ISR里应答的最小副本数大于等于2(min.insync.replicas=2) 消费者 事务+手动提交offset(enable.auto.commit=false) 消费者输出的目的地必须支持事务(Mysql,kafka) 创建一个只有1个分区的topic 测试这个topic的producer和consumer吞吐量 假设他们分别为Tp和Tc，单位是MB/s 假设总的目标吞吐量为Tt，那么分区数=Tt/min(Tp,Tc) message.max.bytes：broker接收每批次消息最大值，默认1m
如果生产环境某个kafka节点挂了：
先尝试重启 增加内存，cpu，带宽 如果将整个kafka节点误删除，如果副本数大于等于2，可按照服役新节点的方式重新服役一个新节点，并执行负载均衡。 kafka压测：官方自带脚本 producer压测 record-size：一条信息大小，单位字节，测试为1K num-records：共发送消息数，测试为100万 throughput是每秒发送消息数，-1表示不限流，测试为1万/s producer-props：配置生产者相关参数：batch.size为16K consumer压测 –bootstrap：kafka集群地址 –topic：topic名称 –messages：总共消费消息数，100万 http://kafka.apache.org/downloads
