前言

什么是EFAK

EFAK（Eagle For Apache Kafka，以前称为 Kafka Eagle）EFAK是开源可视化和管理软件。可以查询、可视化、监控kafka集群，是将 kafka 的集群数据转换为图形可视化的工具。

为什么要用EFAK

• Apache Kafka 没有正式提供监控系统或页面。

• 开源Kafka监控系统功能太少或暂停维护。

• 现有的监控系统难以配置和使用。

• 一些监控系统无法满足与现有IM的集成，如微信、钉钉等。

安装

下载

可以在 GitHub 上下载EFAK源代码自行编译安装，也可以下载二进制 .tar.gz 文件。

EFAK	存储库
Github	https://github.com/smartloli/EFAK
下载	http://download.kafka-eagle.org/

ps：推荐使用官方编译的二进制安装包

「安装JDK」

如果Linux服务器上有JDK环境，这一步可以忽略，继续下一步的安装。如果没有JDK，请先到Oracle官网下载JDK。

JAVA_HOME 配置 将二进制安装包解压到指定目录：

cd /usr/java
tar -zxvf jdk-xxxx.tar.gz
mv jdk-xxxx jdk1.8
vi /etc/profile

export JAVA_HOME=/usr/java/jdk1.8
export PATH=$PATH:$JAVA_HOME/bin

然后，我们使用. /etc/profile使配置立即生效。

「提取 EFAK」

这里我们解压到/data/soft/new目录并解压：

tar -zxvf efak-xxx-bin.tar.gz

如果之前安装过版本，删除修改后的版本，重命名当前版本，如下图：

rm -rf efak
mv efak-xxx efak

然后，配置 EFAK 配置文件

vi /etc/profileexport 
KE_HOME=/data/soft/new/efakexport 
PATH=$PATH:$KE_HOME/bin

最后，我们使用. /etc/profile使配置立即生效。

配置EFAK系统文件 根据自身Kafka集群的实际情况配置EFAK，例如zookeeper地址、Kafka集群的版本类型（zk为低版本，kafka为高版本）、开启安全认证的Kafka集群等。

cd ${KE_HOME}/conf
vi system-config.properties

# Multi zookeeper&kafka cluster list -- The client connection address of the Zookeeper cluster is set here
efak.zk.cluster.alias=cluster1,cluster2
cluster1.zk.list=tdn1:2181,tdn2:2181,tdn3:2181
cluster2.zk.list=xdn1:2181,xdn2:2181,xdn3:2181

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

# Kafka broker nodes online list
cluster1.efak.broker.size=10
cluster2.efak.broker.size=20

# Zkcli limit -- Zookeeper cluster allows the number of clients to connect to
# If you enable distributed mode, you can set value to 4 or 8
kafka.zk.limit.size=16

# EFAK webui port -- WebConsole port access address
efak.webui.port=8048

######################################
# EFAK enable distributed
######################################
efak.distributed.enable=false
# master worknode set status to master, other node set status to slave
efak.cluster.mode.status=slave
# deploy efak server address
efak.worknode.master.host=localhost
efak.worknode.port=8085

# Kafka offset storage -- Offset stored in a Kafka cluster, if stored in the zookeeper, you can not use this option
cluster1.efak.offset.storage=kafka
cluster2.efak.offset.storage=kafka

# Whether the Kafka performance monitoring diagram is enabled
efak.metrics.charts=false

# EFAK keeps data for 30 days by default
efak.metrics.retain=30

# If offset is out of range occurs, enable this property -- Only suitable for kafka sql
efak.sql.fix.error=false
efak.sql.topic.records.max=5000

# Delete kafka topic token -- Set to delete the topic token, so that administrators can have the right to delete
efak.topic.token=keadmin

# Kafka sasl authenticate
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";
# If not set, the value can be empty
cluster1.efak.sasl.client.id=
# Add kafka cluster cgroups
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=kafka_ads01,kafka_ads02

cluster2.efak.sasl.enable=true
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";
cluster2.efak.sasl.client.id=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=kafka_ads03,kafka_ads04

# Default use sqlite to store data
efak.driver=org.sqlite.JDBC
# It is important to note that the '/hadoop/kafka-eagle/db' path must be exist.
efak.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
efak.username=root
efak.password=smartloli

# (Optional) set mysql address
#efak.driver=com.mysql.jdbc.Driver
#efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
#efak.username=root
#efak.password=smartloli

「启动EFAK服务器（独立）」 在$KE_HOME/bin目录中，有一个ke.sh脚本文件。执行启动命令如下：

cd ${KE_HOME}/bin
chmod +x ke.sh ke.sh start

之后，当 EFAK 服务器重新启动或停止时，执行以下命令：

ke.sh restart
ke.sh stop

如下图所示：

「启动EFAK服务器（分布式）」

在$KE_HOME/bin目录中，有一个ke.sh脚本文件。执行启动命令如下：

cd ${KE_HOME}/bin
# sync efak package to other worknode node
# if $KE_HOME is /data/soft/new/efak
for i in `cat $KE_HOME/conf/works`;do scp -r $KE_HOME $i:/data/soft/new;done

# sync efak server .bash_profile environment
for i in `cat $KE_HOME/conf/works`;do scp -r ~/.bash_profile $i:~/;done

chmod +x ke.sh 
ke.sh cluster start

之后，当 EFAK 服务器重新启动或停止时，执行以下命令：

ke.sh cluster restart
ke.sh cluster stop

如下图所示：

使用

仪表盘

查看Kafka brokers、topics、consumer 、Zookeepers信息等

创建主题

列出主题

该模块跟踪Kafka集群中的所有Topic，包括分区数、创建时间、修改Topic，如下图所示：

主题详情

每个Topic对应一个超链接，可以查看Topic的详细信息，如下图：

消费情况

数据预警

配置下邮件服务器的用户名和密码，就可以看到相应的预警数据了

以及最后的大屏展示

以上就是EFAK的安装和使简单用了，具体的项目里还有很多功能，推荐大家可以使用EFAK作为Kafka的可视化管理软件

扫码关注我的微信公众号：Java架构师进阶编程  获取最新面试题，电子书

专注分享Java技术干货，包括JVM、SpringBoot、SpringCloud、数据库、架构设计、面试题、电子书等，期待你的关注！