广西藤县疫情_软件开发外包公司赚钱不_app拉新推广一手接单平台_做引流的公司是正规的吗

Kafka 在 ZooKeeper 里面的存储结构

topic 结构

/brokers/topics/[topic]

partition结构

/brokers/topics/[topic]/partitions/[partitionId]/state

broker信息

/brokers/ids/[o...N]

控制器

/controller
存储center controller中央控制器所在kafka broker的信息

消费者

/consumers/[groupId]/ids /[consumerIdstring]
每个consumer都有一个唯一的ID,此id用来标记消费者信息
消费者管理者:
/consumers/[groupId]/owners/[topic]/[partitionid]

图形工具 Offset Explorer (Kafka Tool)

Offset Explorer ，旧称Kafka Tool，工具是一个 GUI 应用程序，用于管理和使用 Apache Kafka 群集。它提供了一个直观的 UI，允许人们快速查看 Kafka 群集中的对象以及存储在群集主题中的消息。它包含面向开发人员和管理员的功能。一些关键功能包括

快速查看您的所有 Kafka 集群，包括其经纪人、主题和消费者
查看分区中邮件的内容并添加新邮件
查看消费者的偏移量，包括阿帕奇风暴卡夫卡喷口消费者
以漂亮的打印格式显示 JSON和 XML 消息
添加和删除主题以及其他管理功能
将单个邮件从分区保存到本地硬盘驱动器
编写自己的插件，允许您查看自定义数据格式
Kafka 工具在Windows、Linux 和 Mac 操作系统上运行

官网:

Offset Explorer

下载链接:

Offset Explorer

>^C[root@node1 ~]#/usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.187:9092 --topic kun
>hello1
>

查看数据

基于Web的Kafka集群监控系统 kafka-eagle

介绍

Kafka eagle（kafka鹰）是一款由国内公司开源的Kafka集群监控系统，可以用来监视kafka集群的broker状态、Topic信息、IO、内存、consumer线程、偏移量等信息，并进行可视化图表展示。独特的KQL还可以通过SQL在线查询kafka中的数据。

官方地址

http://www.kafka-eagle.org/
https://github.com/smartloli/kafka-eagle-bin
https://www.cnblogs.com/smartloli/

安装

安装说明

https://docs.kafka-eagle.org/2.installation
https://www.cnblogs.com/smartloli/p/16728995.html

安装 JAVA

apt update && apt -y install openjdk-8-jdk

下载安装

wget https://github.com/smartloli/kafka-eagle-bin/archive/refs/tags/v3.0.2.tar.gz

解压安装包

tar zxf kafka-eagle-bin-3.0.2.tar.gz
cd kafka-eagle-bin-3.0.2/
tar -zxvf kafka-eagle-web-3.0.2-bin.tar.gz -C /usr/local/
ln -s /usr/local/kafka-eagle-web-3.0.2 /usr/local/kafka-eagle-web

设置全局变量

设置相关全局变量KE_HOME

vi /etc/profile
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export KE_HOME=/usr/local/kafka-eagle-web
export PATH=$PATH:$KE_HOME/bin
. /etc/profile

修改配置文件

vim /usr/local/kafka-eagle-web/conf/system-config.properties
###################################### 
# 填写 zookeeper集群列表
#kafkazookeeper节点配置属性多个可以添加一个cluster1,如果有多套kafka集群加多个名称
efak.zk.cluster.alias=cluster1,cluster2 
######################################
#zookeeper地址
######################################修改此处
cluster1.zk.list=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181
#cluster2.zk.list=10.0.0.201:2181,10.0.0.202:2181,10.0.0.203:2181
######################################
# broker 最大规模数量
###################################### 
cluster1.efak.broker.size=20
######################################
# zk 客户端线程数#####################
################ 
kafka.zk.limit.size=32
######################################
# EFAK webui 端口
######################################
fak.webui.port=8048
####################################
# kafka offset storage 
###################################### 
cluster1.efak.offset.storage=kafka 
cluster2.efak.offset.storage=zk
######################################
# kafka jmx uri 
######################################
cluster1.efak.jmx.uri=service:jmx:rmi:///jn di/rmi://%s/jmxrmi
######################################
# kafka metrics 指标，默认存储15天
###################################### 
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 sqlite 数据库地址（需要修改存储路径） 
######################################修改此处,取消下面四行注释
efak.driver=org.sqlite.JDBC
efak.url=jdbc:sqlite:/usr/local/kafka-eagle-web/db/ke.db
efak.username=root
efak.password=www.kafka-eagle.org
######################################
# kafka mysql 数据库地址（需要提前创建ke库） 
######################################修改此处添加注释下面四行 
#efak.driver=com.mysql.cj.jdbc.Driver 
#efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF8&zeroDateTimeBehavior=convertToNull 
#efak.username=ke
#efak.password=123456

配置文件参考

[root@node1 ~]#cat /usr/local/kafka-eagle-web/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=10.0.0.187:2181,10.0.0.188:2181,10.0.0.189:2181
#cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12: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:/usr/local/kafka-eagle-web/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://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
#efak.username=root
#efak.password=123456

启动

/usr/local/kafka-eagle-web/bin/ke.sh start

修改和监控 Kafka

#所有kafka节点修改配置
[root@node1 ~]#vim /usr/local/kafka/bin/kafka-server-start.sh
......
if[ " x$KAFKA_HEAP_OPTS"="x"] ; thenexport KAFKA_HEAP_OPTS=" -Xmx1G-Xms1G"  export JMX_PORT="9999" #添加此行
fi
......
[root@node1 ~]#systemctl restart kafka