当前位置：首页 > news >正文

centos7安装kafka、zookeeper

news 来源：原创 2024/5/14 2:34:50

安装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:2181

broker.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/bin

source /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:2181efak.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

我这边启动报错，根据提升是内存不够，我们把内存调小一点