安装jdk
安装jdk8
安装zookeeper
在指定目录执行下载命令 我是在/newdisk/zookeeper目录下
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz --no-check-certificate
下载好后并解压 tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
校验是否安装好，进入 /newdisk/zookeeper/apache-zookeeper-3.5.8-bin/conf
复制zoo_sample.cfg配置文件为zoo.cfg
cp zoo_sample.cfg zoo.cfg在进入/newdisk/zookeeper/apache-zookeeper-3.5.8-bin/bin目录
启动zk服务端
./zkServer.sh start
启动客户端
./zkCli.sh在查看根节点 ，出现如下界面就安装成功了
安装kafka
进入到要下载的目录，我这边是/newdisk/kafka 执行下载命令（有点慢）
wget https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgz --no-check-certificate或者直接使用安装包kafka_2.13-3.6.0.tgz
解压 tar -xvzf kafka_2.13-3.6.0.tgz
修改配置
进入/newdisk/kafka/kafka_2.13-3.6.0/config，vim server.properties
# broker.id属性在kafka集群中必须要是唯一
broker.id=0
# kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://内网ip:9092
# kafka的消息存储文件
log.dirs=/usr/local/data/kafka-logs
# kafka连接zookeeper的地址
zookeeper.connect=内网ip:2181broker.id=0
listeners=PLAINTEXT://192.168.31.131:9092
log.dir=/newdisk/kafka/data/kafka-logs
zookeeper.connect=192.168.31.131:2181启动kafka（前提是zookeeper服务端启动）
进入/newdisk/kafka/kafka_2.13-3.6.0/，执行如下命令
bin/kafka-server-start.sh config/server.properties通过zookeeper客户端查看是否注册成功
kafka基本操作
创建topic
参数 --topic 指定 Topic 名，--partitions 指定分区数，--replication-factor 指定副本数：
./kafka-topics.sh --create --bootstrap-server 192.168.31.131:9092 --replication-factor 1 --partitions 1 --topic test
查看topic
./kafka-topics.sh --list --bootstrap-server 192.168.31.131:9092发送消息
./kafka-console-producer.sh --broker-list 192.168.31.131:9092 --topic test
消费消息
./kafka-console-consumer.sh --bootstrap-server 192.168.31.131:9092 --topic test --from-beginning
安装可视化工具Apache Kafka UI
地址：GitHub - provectus/kafka-ui: Open-Source Web UI for Apache Kafka Management
简介
Apache Kafka UI 是一个免费的开源 Web UI，用于监控和管理 Apache Kafka 集群。
Apache Kafka UI 是一个简单的工具，可让您的数据流可观察，帮助更快地发现和解决问题并提供最佳性能。其轻量级仪表板可以轻松跟踪 Kafka 集群的关键指标 - 代理、主题、分区、生产和消耗。
安装及启动（docker）
运行如下命令
docker run -it -p 8080:8080 -d -e DYNAMIC_CONFIG_ENABLED=true provectuslabs/kafka-ui然后访问 Web UI ：http://localhost:8080
可视化工具kafka-eagle（推荐）
下载 ，只需要到自己想要下载的文件路径下执行如下命令
wget https://github.com/smartloli/kafka-eagle-bin/archive/v2.1.0.tar.gz
解压，解压后发现里面还是一个压缩包
tar -xzvf v2.1.0.tar.gz
继续解压
tar -xvzf efak-web-2.1.0-bin.tar.gz复制到指定文件夹
mv efak-web-2.1.0 /newdisk/kafka/kafka-web
配置kafka_eagle环境变量：vim /etc/profile
# KAFKA_EAGLE
export KE_HOME=/newdisk/kafka/kafka-web
export PATH=$PATH:$KE_HOME/binsource /etc/profile
配置kafka_eagle，修改配置文件conf文件夹下的system-config.properties文件
完整的配置如下
######################################
# multi zookeeper & kafka cluster list
# Settings prefixed with 'kafka.eagle.' will be deprecated, use 'efak.' instead
######################################
#efak.zk.cluster.alias=cluster1,cluster2
#cluster1.zk.list=tdn1:2181,tdn2:2181,tdn3:2181
#cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181
efak.zk.cluster.alias=cluster1
cluster1.zk.list=192.168.31.131:2181
######################################
# 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.efak.broker.size=20
######################################
# zk client thread limit
######################################
kafka.zk.limit.size=16
######################################
# EFAK webui port
######################################
efak.webui.port=8048
######################################
# EFAK enable distributed
######################################
efak.distributed.enable=false
efak.cluster.mode.status=master
efak.worknode.master.host=localhost
efak.worknode.port=8085
######################################
# kafka jmx acl and ssl authenticate
######################################
cluster1.efak.jmx.acl=false
cluster1.efak.jmx.user=keadmin
cluster1.efak.jmx.password=keadmin123
cluster1.efak.jmx.ssl=false
cluster1.efak.jmx.truststore.location=/data/ssl/certificates/kafka.truststore
cluster1.efak.jmx.truststore.password=ke123456
######################################
# kafka offset storage
######################################
cluster1.efak.offset.storage=kafka
#cluster2.efak.offset.storage=zk
######################################
# kafka jmx uri
######################################
cluster1.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi
######################################
# kafka metrics, 15 days by default
######################################
efak.metrics.charts=true
efak.metrics.retain=15
######################################
# kafka sql topic records max
######################################
efak.sql.topic.records.max=5000
efak.sql.topic.preview.records.max=10
######################################
# delete kafka topic token
######################################
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="kafka" password="kafka-eagle";
cluster1.efak.sasl.client.id=
cluster1.efak.blacklist.topics=
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=
#cluster2.efak.sasl.enable=false
#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="kafka" password="kafka-eagle";
#cluster2.efak.sasl.client.id=
#cluster2.efak.blacklist.topics=
#cluster2.efak.sasl.cgroup.enable=false
#cluster2.efak.sasl.cgroup.topics=
######################################
# kafka ssl authenticate
######################################
cluster3.efak.ssl.enable=false
cluster3.efak.ssl.protocol=SSL
cluster3.efak.ssl.truststore.location=
cluster3.efak.ssl.truststore.password=
cluster3.efak.ssl.keystore.location=
cluster3.efak.ssl.keystore.password=
cluster3.efak.ssl.key.password=
cluster3.efak.ssl.endpoint.identification.algorithm=https
cluster3.efak.blacklist.topics=
cluster3.efak.ssl.cgroup.enable=false
cluster3.efak.ssl.cgroup.topics=
######################################
# kafka sqlite jdbc driver address
######################################
#efak.driver=org.sqlite.JDBC
#efak.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
#efak.username=root
#efak.password=www.kafka-eagle.org
######################################
# kafka mysql jdbc driver address
######################################
efak.driver=com.mysql.cj.jdbc.Driver
efak.url=jdbc:mysql://192.168.31.131:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
efak.username=root
efak.password=123456
保存后，启动
bin/ke.sh start
我这边启动报错，根据提升是内存不够，我们把内存调小一点
vim bin/ke.sh
重新启动
查看状态
bin/ke.sh status
浏览器访问http://192.168.31.131:8048/ 用户名/密码 admin/123456
到这边就成功了
![[计算机网络]：流量控制](https://img-blog.csdnimg.cn/direct/66488ccfb58c44a39489776e59ada41d.png)
