做网站接私活价格怎么算,百度小程序app下载,做代码的网站,建设网站架构服务器需要多少钱前言
ZooKeeper 是一个分布式的、开放源码的分布式应用程序协调服务#xff0c;提供配置维护、命名服务、分布式同步、组服务等功能#xff1b;
Kafka 是一个开源的分布式流处理平台#xff0c;它被设计用来处理实时数据流#xff0c;包括发布和订阅消息系统、日志收集以… 前言
ZooKeeper 是一个分布式的、开放源码的分布式应用程序协调服务提供配置维护、命名服务、分布式同步、组服务等功能
Kafka 是一个开源的分布式流处理平台它被设计用来处理实时数据流包括发布和订阅消息系统、日志收集以及作为事件流数据平台
在 Kafka 集群中ZooKeeper 用于协调和管理 Kafka broker 的状态、集群的配置信息以及其他关键元数据。结合使用时ZooKeeper 提供了 Kafka 所需的集群管理和协调功能使得 Kafka 集群更加稳定可靠同时也简化了 Kafka 集群的管理和运维。
目录
一、ZooKeeper 服务
1. 简介
2. 工作机制
3. 特点
4. Zookeeper 数据结构
5. 应用场景
6. 选举机制
6.1 第一次启动选举机制
6.2 非第一次启动选举机制
6.2.1 Leader 选举两种情况
6.2.2 Leader 选举流程时状态
二、部署 Zookeeper 集群
1. 环境准备
2. 下载 zookeeper 安装包
3. 安装 zookeeper
4. 创建数据目录和日志目录
5. 创建 myid 文件
6. 配置 Zookeeper 启动脚本
7. 分别启动 Zookeeper
三、Kafka 消息队列
1. 概述
2. 使用消息队列的好处
2.1 解耦
2.2 可恢复性
2.3 缓冲
2.4 灵活性及峰值处理能力
2.5 异步通信
3. 消息队列的两种模式
3.1 点对点模式
3.2 发布/订阅模式
4. Kafka 的特性
5. Kafka 系统架构
5.1 架构介绍
5.2 Partation 数据路由规则
5.3 分区原因
四、部署 Kafka 集群
1. 环境准备
2. 下载 Kafka 安装包
3. 安装 Kafka
4. 配置 Zookeeper 启动脚本
5. 分别启动 Kafka
6. Kafka 命令行操作
7. 创建主题一般故障
五、KafkaELKF
1. 修改 filebeat 的主配置文件
2. 在 filebeat 节点安装 apache
3. 新建一个 Logstash kafka 配置文件
4. 查看所有的索引
5. 登录 Kibana 添加索引 一、ZooKeeper 服务
1. 简介
Zookeeper 是一个开源的分布式的为分布式框架提供协调服务的 Apache 项目。
2. 工作机制
是一个基于观察者模式设计的分布式服务管理框架它负责存储和管理大家都关心的数据然后接受观察者的注册一旦这些数据的状态发生变化Zookeeper 就将负责通知已经在 Zookeeper 上注册的那些观察者做出相应的反应。也就是说 Zookeeper 文件系统 通知机制。
3. 特点
① 一个领导者多个跟随着组成的集群
② 集群中只要有半数以上存活就可以正常服务适合安装奇数台服务器最少3台
③ 全局数据一致每个 server 保存一份相同的数据副本client 无论连接到哪个 server数据都是一致的
④ 更新请求顺序执行来自同一个 client 的更新请求按其发送顺序依次执行即先进先出
⑤ 数据更新原子性一次数据更新要么成功要么失败
⑥ 实时性在一定时间范围内client 能读到最新数据
4. Zookeeper 数据结构 ZooKeeper 数据模型的结构与 Linux 文件系统很类似整体上可以看作是一棵树每个节点称做一个 ZNode。每一个 ZNode 默认能够存储1MB的数据每个 ZNode 都可以通过其路径唯一标识。
当 Kafka 与 ZooKeeper 集成时Kafka 会使用 ZooKeeper 来存储其元数据和配置信息。如果用户与 Kafka 断开连接Kafka 可能会丢失对 ZooKeeper 中特定路径下数据的访问权限或连接状态。当用户再次连接时Kafka 可以利用先前定义的路径来重新建立与 ZooKeeper 的连接并恢复先前的状态以便继续正常运行。
5. 应用场景
① 统一命名服务
在分布式环境中应用/服务需要统一的命名以便于识别。例如将IP地址转换为易记的域名
② 统一配置管理
配置同步在分布式环境下要求集群中所有节点的配置信息是一致的如 Kafka 集群。通过ZooKeeper配置信息可以被写入一个 Znode并且各个客户端服务器可以监听这个 Znode。任何对 Znode 中数据的修改都会被 ZooKeeper 通知到各个客户端服务器快速同步修改后的配置信息能够快速同步到各个节点上
③ 统一集群管理
实时状态监控在分布式环境中实时了解每个节点的状态是必要的因为可以根据节点状态做出调整。ZooKeeper 可以实现实时监控节点状态变化将节点信息写入 ZooKeeper 上的一个 ZNode并监听这个 ZNode 以获取实时状态变化
④ 服务器动态上下线
ZooKeeper 可以让客户端实时感知服务器的上下线变化
⑤ 软负载均衡
在 ZooKeeper 中记录每台服务器的访问数从而让访问数最少的服务器去处理最新的客户端请求
6. 选举机制
在 ZooKeeper 中选举是通过投票机制来实现的。每个服务器都有权投票给自己和其他服务器并且一旦某个服务器获得超过半数的选票它就会成为新的 Leader。
6.1 第一次启动选举机制
服务器启动票数myid状态角色 server1 server11 1 looking server2 server10 server22 2 looking server3 server10 server20 server23 3 server1follower server2follower server3leader server4 server10 server20 server33 server41 4 server1follower server2follower server3leader server4follower server5 server10 server20 server33 server41 server51 5 server1follower server2follower server3leader server4follower server5follower
① server1 启动发起一次选举。投自己一票。此时服务器1为一票不够半数以上3票选举无法完成server1 状态保持为 looking
② server2 启动再发起一次选举。server1、server2 各投自己一票并交换选票信息此时server1 发现 server2 的 myid 比自己目前投票推举的server1大更改选票为推举 server2。此时server1 票数0票server2 票数2票没有半数以上结果选举无法完成server1、server2 状态保持 looking
③ server3 启动发起一次选举。此时server1、server2 都会更改选票为 server3。此次投票结果server1 为0票server2 为0票server3 为3票。此时server3 的票数已经超过半数server3当选 leader。server1、server2 更改状态为 followerserver3 更改状态为 leader
④ server4 启动发起一次选举。此时server123已经不是 looking 状态不会更改选票信息。交换选票信息结果server3 为3票server4 为1票。此时server4 服从多数更改选票信息为server3并更改状态为 follower
⑤ server5 启动同 server4 一样为 follower。
6.2 非第一次启动选举机制
6.2.1 Leader 选举两种情况
当 ZooKeeper 集群中的一台服务器出现以下两种情况之一时就会开始进入Leader选举
① 服务器初始化启动
② 服务器运行期间无法和 Leader 保持连接
6.2.2 Leader 选举流程时状态
当一台机器进入 Leader 选举流程时当前集群也可能会处于以下两种状态
① 集群中本来就已经存在一个 Leader
仅需要和 Leader 机器建立连接并进行状态同步即可
② 集群中确实不存在 Leader
假设 ZooKeeper 由5台服务器组成SID唯一标识符通常也称为 myid分别为1、2、3、4、5ZXID事务ID分别为8、8、8、7、7并且此时 SID 为3的服务器是Leader。此时3和5服务器出现故障因此开始进行 Leader 选举。 在进行 Leader 选举时按照以下规则进行
如果EPOCH值大的服务器直接胜出如果EPOCH相同比较事务IDZXID事务ID大的胜出如果EPOCH和事务ID都相同则比较服务器IDSID服务器ID大的胜出
在ZooKeeper中Epoch时代是用来标识一轮Leader选举过程的编号。每当进行一次新的Leader选举时Epoch会递增用于区分不同的选举轮次。由于没有 Leader 时同一轮投票过程中的逻辑时钟值是相同的所以不存在 EPOCH 最大值同时也不存在 ZXID 最大值。因此该示例最终 server4 成为新的 Leader。
二、部署 Zookeeper 集群
1. 环境准备 服务器ip 节点名称myid/角色软件版本jdk版本操作系统192.168.190.104zk-kfk011/followzookeeper-3.5.7openjdk version 1.8.0_131centos7.4192.168.190.105zk-kfk022/followzookeeper-3.5.7openjdk version 1.8.0_131centos7.4192.168.190.106zk-kfk033/leaderzookeeper-3.5.7openjdk version 1.8.0_131centos7.4
systemctl stop firewalld.service
setenforce 0
[rootlocalhost ~]# java -version
openjdk version 1.8.0_131 # 自带java环境openjdk# 编辑域名解析制作映射加快传输速度
echo 192.168.190.104 zk-kfk01 /etc/hosts
echo 192.168.190.105 zk-kfk02 /etc/hosts
echo 192.168.190.106 zk-kfk03 /etc/hosts
2. 下载 zookeeper 安装包
三台节点一起配置
[rootlocalhost ~]# cd /opt/
[rootlocalhost opt]# wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.7/apache-zookeeper-3.5.7-bin.tar.gz
[rootlocalhost opt]# ls
apache-zookeeper-3.5.7-bin.tar.gz
3. 安装 zookeeper
三台节点一起配置
[rootlocalhost opt]# tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz
[rootlocalhost opt]# mv apache-zookeeper-3.5.7-bin /usr/local/zookeeper-3.5.7
[rootlocalhost opt]# cd /usr/local/zookeeper-3.5.7/conf/
[rootlocalhost conf]# ls
configuration.xsl log4j.properties zoo_sample.cfg
[rootlocalhost conf]# cp zoo_sample.cfg zoo.cfg
[rootlocalhost conf]# vim zoo.cfg # 修改配置文件2 tickTime2000 # 通信心跳时间Zookeeper服务器与客户端心跳时间单位毫秒5 initLimit10 # Leader和Follower初始连接时能容忍的最多心跳数tickTime的数量这里表示为10*2s8 syncLimit5 # Leader和Follower之间同步通信的超时时间这里表示如果超过5*2sLeader认为Follwer死掉并从服务器列表中删除Follwer12 dataDir/usr/local/zookeeper-3.5.7/data # 修改指定保存Zookeeper中的数据的目录目录需要单独创建13 dataLogDir/usr/local/zookeeper-3.5.7/logs # 添加指定存放日志的目录目录需要单独创建15 clientPort2181 # 客户端连接端口30 server.1192.168.190.104:3188:328831 server.2192.168.190.105:3188:328832 server.3192.168.190.106:3188:3288
server.AB:C:D
# A是一个数字表示这个是第几号服务器。集群模式下需要在zoo.cfg中dataDir指定的目录下创建一个文件myid这个文件里面有一个数据就是A的值Zookeeper启动时读取此文件拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。
# B是这个服务器的地址。
# C是这个服务器Follower与集群中的Leader服务器交换信息的端口。
# D是万一集群中的Leader服务器挂了需要一个端口来重新进行选举选出一个新的Leader而这个端口就是用来执行选举时服务器相互通信的端口。4. 创建数据目录和日志目录
三台节点一起配置
[rootlocalhost ~]# mkdir /usr/local/zookeeper-3.5.7/data
[rootlocalhost ~]# mkdir /usr/local/zookeeper-3.5.7/logs5. 创建 myid 文件
在每个节点的 dataDir 指定的目录下创建一个 myid 的文件dataDir 是用于存储 ZooKeeper 服务器数据的目录。
[rootzk-kfk01 ~]# echo 1 /usr/local/zookeeper-3.5.7/data/myid
[rootzk-kfk02 ~]# echo 2 /usr/local/zookeeper-3.5.7/data/myid
[rootzk-kfk03 ~]# echo 3 /usr/local/zookeeper-3.5.7/data/myid6. 配置 Zookeeper 启动脚本 三台节点一起配置
vim /etc/init.d/zookeeper
#!/bin/bash
#chkconfig:2345 20 90 # 在运行级别2、3、4、5下在启动优先级顺序为20在关闭顺序为90。
#description:Zookeeper Service Control Script
ZK_HOME/usr/local/zookeeper-3.5.7 # 定义了 ZooKeeper 的安装路径
case $1 in # 开始一个 case 语句根据传入的参数进行不同的操作
start) # 如果参数是 start则执行 ZooKeeper 启动命令echo ---------- zookeeper 启动 ------------ $ZK_HOME/bin/zkServer.sh start
;;
stop) # 如果参数是 stop则执行 ZooKeeper 停止命令echo ---------- zookeeper 停止 ------------$ZK_HOME/bin/zkServer.sh stop
;;
restart) # 如果参数是 restart则执行 ZooKeeper 重启命令。echo ---------- zookeeper 重启 ------------$ZK_HOME/bin/zkServer.sh restart
;;
status) # 如果参数是 status则执行 ZooKeeper 查看状态命令echo ---------- zookeeper 状态 ------------$ZK_HOME/bin/zkServer.sh status
;;
*) # 如果参数不匹配以上任何情况则显示使用说明echo Usage: $0 {start|stop|restart|status}
esac
# 管理ZooKeeper服务通过传入不同的参数来启动、停止、重启和查看状态
7. 分别启动 Zookeeper
[rootzk-kfk01 ~]# chmod x /etc/init.d/zookeeper
[rootzk-kfk01 ~]# chkconfig --add zookeeper
# 将 zookeeper 服务添加到系统的服务管理列表中并且配置它在系统启动时自动运行
# 前提创建一个名为 zookeeper 的服务脚本通常是放在 /etc/init.d/ 目录下依次启动不要一起启动
[rootzk-kfk01 ~]# service zookeeper start
[rootzk-kfk02 ~]# service zookeeper start
[rootzk-kfk03 ~]# service zookeeper start
---------- zookeeper 启动 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.5.7/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED[rootzk-kfk01 ~]# service zookeeper status # 查看当前状态
---------- zookeeper 状态 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper-3.5.7/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower三、Kafka 消息队列
1. 概述
Kafka 即消息队列简称 MQ。在高并发环境下当同时处理大量请求访问服务器往往请求会发生阻塞。如大量访问数据库导致锁表引发雪崩效应使用消息队列可以异步处理请求流量消峰、应用解耦从而缓解压力。当 Kafka 组合集群每秒可以处理几十万条数据请求大大增加了抗高并发能力。
当前比较常见的 MQ 中间件有 ActiveMQ淘汰、RabbitMQ用的多、RocketMQ企业常用、Kafka大数据分析、大量数据收集 等。
2. 使用消息队列的好处
2.1 解耦
通过消息队列来实现系统内部各个组件或模块之间的解耦合即两边遵守约定则允许通信。
假设有一个电子商务平台其中订单系统、库存系统和物流系统需要相互通信。通过消息队列解耦它们可以实现以下功能
订单下单当顾客下单时订单系统将订单信息发布到消息队列中而不需要直接调用库存系统或物流系统的接口库存更新库存系统订阅了订单系统发布的消息队列一旦接收到订单信息它会检查库存并更新库存状态物流处理同时物流系统也订阅了订单系统发布的消息队列一旦接收到订单信息它会开始安排物流配送
这种架构下各个系统之间并不直接依赖于彼此的接口而是通过消息队列进行松耦合的通信。
2.2 可恢复性
通过消息队列即使某个处理消息的组件出现故障系统仍然可以保持稳定运行。消息会被安全地存储在队列中等待故障组件重新上线后进行处理从而提高了整个系统的可靠性。
2.3 缓冲
消息队列可以作为缓冲层帮助控制系统中数据流的速度。
2.4 灵活性及峰值处理能力
消息队列可以帮助系统应对突发的访问量增加确保关键组件能够应对压力而不至于崩溃。这提供了系统在面对异常情况下的弹性和稳定性。
2.5 异步通信 允许用户将消息放入队列而不需要立即处理可以提高系统的响应速度和整体效率同时也降低了对实时处理的需求。
3. 消息队列的两种模式
3.1 点对点模式 即一对一消息收到后消息清除。在点对点模式中消息生产者将消息发送到队列中然后消息消费者从队列中主动拉取并处理消息。一旦消息被某个消费者处理完毕它就会从队列中清除因此同一条消息只能被一个消费者处理。这种模式适用于需要确保每条消息只被处理一次的场景。
3.2 发布/订阅模式 即一对多又叫观察者模式消费者消费数据之后不会清除消息。观察者的作用是接收消息队列中特定主题或频道上的消息并通知消费者或其他观察者。在发布/订阅模式中消息生产者将消息发布到特定的主题topic中而多个消息消费者可以订阅这个主题并独立地消费消息。与点对点模式不同发布到主题的消息会被所有订阅者接收和消费。这种模式适用于需要将消息广播给多个订阅者的场景例如直播。
4. Kafka 的特性
① 高吞吐量、低延迟
Kafka 每秒可以处理几十万条消息它的延迟最低只有几毫秒。每个 topic 可以分多个 PartitionConsumer Group 对 Partition 进行消费操作提高负载均衡能力和消费能力
② 可扩展性
kafka 集群支持热扩展
③ 持久性、可靠性
消息被持久化到本地磁盘并且支持数据备份冗余
④ 容错性
允许集群中节点失败多副本情况下若副本数量为 n则允许 n-1 个节点失败
⑤ 高并发
支持数千个客户端同时读写
5. Kafka 系统架构 5.1 架构介绍
① Broker 服务器
Broker 服务器是 Kafka 中的服务器单元每个 Broker 可以理解为一个 Kafka 实例负责存储和管理消息。一台 Kafka 服务器就是一个 Broker一个集群由多个 Broker 组成一个 Broker 可以容纳多个 Topic。
② Topic 主题
主题是消息的逻辑容器类似于消息队列中的队列。生产者发布消息到主题消费者从主题订阅消息。每个主题可以划分为多个分区以便实现数据的分布和负载均衡。
③ Partition 分区
为了实现扩展性和高吞吐量一个主题可以分割为多个分区每个分区是一个有序的队列。分区允许主题的数据分布到多个 broker 上并且提供了并行处理消息的能力。Kafka 保证每个分区内的消息顺序但不保证不同分区之间的消息顺序。或者说 Kafka 只保证 Partition 内的记录是有序的而不保证 Topic 中不同 Partition 的顺序。
每个 Topic 至少有一个 Partition当生产者产生数据的时候会根据分配策略选择分区然后将消息追加到指定的分区的队列末尾。
5.2 Partation 数据路由规则
① 指定了 Patition如果消息的生产者明确指定了分区Partition则消息将被直接发送到该分区中
② 未指定 Patition 但指定 Key相当于消息中某个属性如果消息的生产者没有指定分区但指定了一个键Key则根据该键的值进行哈希运算并取模以确定消息应该发送到的分区
③ Patition 和 Key 都未指定如果消息的发送者既没有指定分区也没有指定键则系统将根据轮询的方式从可用的分区中选取一个分区。 关于分区的一些补充信息
消息偏移量编号 每条消息都有一个自增的编号用于标识消息的偏移量从0开始标识顺序。分区中的数据存储方式 每个分区使用多个 Segment 文件来存储数据。严格保证消息消费顺序的情况下 如果有严格保证消息消费顺序的需求比如商品秒杀、抢红包等可以将分区数目设置为1这样可以保证消息在消费时的顺序性。 关于 Broker 存储数据的情况
Broker 存储 Topic 数据 如果某个 Topic 有 N 个分区那么集群中的每个 Broker 存储该 Topic 的一个分区。Broker 数量与 Topic 分区数的关系 如果某个 Topic 有 N 个分区而集群中有 NM 个 Broker那么其中的 N 个 Broker 分别存储该 Topic 的一个分区而剩下的 M 个 Broker 不存储该 Topic 的分区数据。Broker 数量少于 Topic 分区数的情况 如果某个 Topic 有 N 个分区但集群中的 Broker 数量少于 N 个那么一个 Broker 可能会存储该 Topic 的一个或多个分区。在实际生产环境中应尽量避免这种情况因为这可能导致 Kafka 集群数据不均衡。
5.3 分区原因
方便在集群中扩展每个 Partition 可以通过调整以适应它所在的机器而一个 Topic 又可以有多个 Partition 组成因此整个集群就可以适应任意大小的数据了以 Partition 为单位读写Kafka 使用分区来方便集群扩展和提高并发性
① Replica 副本
为了防止节点故障导致数据丢失Kafka 提供了副本机制。每个分区都有若干个副本其中一个是 Leader负责读写其他是 Follower负责备份。
② Leader 领导者
每个 Partition 有多个副本其中有且仅有一个作为 LeaderLeader 是当前负责数据的读写的 Partition。
③ Follower 跟随者
Follower 跟随 Leader负责备份数据。如果 Leader 故障会从 Follower 中选举出新的 Leader。
④ producer 生产者
生产者即数据的发布者将消息发布到 Kafka 的 topic 中可以指定数据存储的分区。Broker 接收到生产者发送的消息后Broker 将该消息追加到当前用于追加数据的 Segment通过 Segment.io 收集的数据所存储的文件文件中。
⑤ Consumer 消费者
从 Kafka 中拉取数据进行消费可以消费多个 topic。
⑥ Consumer GroupCG消费者组
多个消费者组成一个消费者组每个组内的消费者负责消费不同分区的数据防止数据被重复读取将多个消费者集中到一起去处理某一个 Topic 的数据可以更快的提高数据的消费能力。
⑦ offset 偏移量
唯一标识一条消息的位置信息消费者通过偏移量确定下次读取消息的位置。消息最终还是会被删除的默认生命周期为 1 周7*24小时。
⑧ Zookeeper
Kafka 使用 Zookeeper 存储集群的元信息例如 broker 的地址和分区的状态同时用于管理消费者组的偏移量。Kafka 0.9 版本之前consumer 默认将 offset 保存在 Zookeeper 中从 0.9 版本开始consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中该 topic 为 __consumer_offsets。
总之zookeeper 的作用就是生产者 push 数据到 kafka 集群就必须要找到 kafka 集群的节点在哪里这些都是通过 zookeeper 去寻找的。消费者消费哪一条数据也需要 zookeeper 的支持从 zookeepe r获得 offsetoffset 记录上一次消费的数据消费到哪里这样就可以接着下一条数据进行消费。
四、部署 Kafka 集群
1. 环境准备 服务器ip 节点名称myid/角色软件版本jdk版本操作系统192.168.190.104zk-kfk011/follow zookeeper-3.5.7 kafka_2.13-2.7.1 openjdk version 1.8.0_131centos7.4192.168.190.105zk-kfk022/follow zookeeper-3.5.7 kafka_2.13-2.7.1 openjdk version 1.8.0_131centos7.4192.168.190.106zk-kfk033/leader zookeeper-3.5.7 kafka_2.13-2.7.1 openjdk version 1.8.0_131centos7.4
2. 下载 Kafka 安装包
三台节点一起操作
官方下载地址http://kafka.apache.org/downloads.html
[rootzk-kfk01 opt]# wget https://archive.apache.org/dist/kafka/2.7.1/kafka_2.13-2.7.1.tgz
[rootzk-kfk01 opt]# ls
kafka_2.13-2.7.1.tgz
3. 安装 Kafka
三台节点一起操作注意 broker.id 和 listeners 地址
[rootzk-kfk01 opt]# tar zxvf kafka_2.13-2.7.1.tgz
[rootzk-kfk01 opt]# mv kafka_2.13-2.7.1 /usr/local/kafka
修改配置文件
[rootzk-kfk01 opt]# cd /usr/local/kafka/config/
[rootzk-kfk01 config]# cp server.properties{,.bak} # 备份配置文件
[rootzk-kfk01 config]# vim server.properties21 broker.id0(12) # broker的全局唯一编号每个broker不能重复另外两台broker.id1、broker.id231 listenersPLAINTEXT://192.168.190.103:9092 104105
# 指定监听的IP和端口如果修改每个broker的IP需区分开来也可保持默认配置不用修改42 num.network.threads3 # broker处理网络请求的线程数量一般情况下不需要去修改45 num.io.threads8 # 用来处理磁盘IO的线程数量数值应该大于硬盘数48 socket.send.buffer.bytes102400 # 发送套接字的缓冲区大小51 socket.receive.buffer.bytes102400 # 接收套接字的缓冲区大小54 socket.request.max.bytes104857600 # 请求套接字的缓冲区大小60 log.dirs/usr/local/kafka/logs # kafka运行日志存放的路径也是数据存放的路径65 num.partitions1 # topic在当前broker上的默认分区个数会被topic创建时的指定参数覆盖69 num.recovery.threads.per.data.dir1 # 用来恢复和清理data下数据的线程数量
103 log.retention.hours168 # segment文件数据文件保留的最长时间单位为小时默认为7天超时将被删除
110 log.segment.bytes1073741824 # 一个segment文件最大的大小默认为 1G超出将新建一个新的segment文件
123 zookeeper.connect192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181
# 配置连接Zookeeper集群地址修改环境变量:
[rootzk-kfk01 ~]# vim /etc/profile
export KAFKA_HOME/usr/local/kafka # 将一个名为 KAFKA_HOME 的环境变量设置为 /usr/local/kafka这表示 Kafka 的安装路径
export PATH$PATH:$KAFKA_HOME/bin # 将 Kafka 的 bin 目录添加到系统的 PATH 环境变量中
[rootzk-kfk01 ~]# source /etc/profile # 重新加载 /etc/profile 文件中设置的环境变量
4. 配置 Zookeeper 启动脚本
三台节点一起操作
[rootzk-kfk01 ~]# vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME/usr/local/kafka # 定义了 Kafka 的安装路径
case $1 in # 开始一个 case 语句根据传入的参数进行不同的操作
start) # 如果参数是 start则执行 Kafka 启动命令echo ---------- Kafka 启动 ------------${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop) # 如果参数是 stop则执行 Kafka 停止命令echo ---------- Kafka 停止 ------------${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart) # 如果参数是 restart则先执行 Kafka 停止命令再执行 Kafka 启动命令$0 stop $0 start
;;
status) # 如果参数是 status则检查 Kafka 进程是否在运行并输出相应的状态信息echo ---------- Kafka 状态 ------------count$(ps -ef | grep kafka | egrep -cv grep|$$)if [ $count -eq 0 ];thenecho kafka is not runningelseecho kafka is runningfi
;;
*) # 如果参数不匹配以上任何情况则显示使用说明echo Usage: $0 {start|stop|restart|status}
esac
# 管理 Kafka 服务可以通过传入不同的参数来启动、停止、重启和查看状态
5. 分别启动 Kafka
[rootzk-kfk02 ~]# chmod x /etc/init.d/kafka
[rootzk-kfk02 ~]# chkconfig --add kafka
依次启动 kafaka:
[rootzk-kfk01 ~]# service kafka start # 启动 Kafka
[rootzk-kfk02 ~]# service kafka start
[rootzk-kfk03 ~]# service kafka start
---------- Kafka 启动 ------------
[rootzk-kfk01 ~]# service kafka status # 查看状态
---------- Kafka 状态 ------------
kafka is running6. Kafka 命令行操作
① 创建一个名为 test 的 Kafka 主题topic任意一台机器即可
[rootzk-kfk01 ~]# kafka-topics.sh --create --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181 --replication-factor 2 --partitions 3 --topic test
# kafka-topics.sh: 这是 Kafka 提供的一个脚本工具用于管理 Kafka 中的主题。
# --create: 这是告诉 kafka-topics.sh 要创建一个新的主题。
# --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181: 这是指定 ZooKeeper 的连接信息用于管理 Kafka 集群。在这里指定了三个 ZooKeeper 实例的地址和端口号。
# --replication-factor 2: 这表示设置每个分区的副本数量为 2。副本是为了数据的冗余和容错性而存在的。
# --partitions 3: 这表示要创建 3 个分区。
# --topic test: 这表示要创建一个名为 test 的主题。
# 在 Kafka 集群中创建一个名为 test 的主题该主题有 3 个分区并且每个分区有 2 个副本。
② 查看当前服务器中的所有 topic
[rootzk-kfk01 config]# kafka-topics.sh --list --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181
test
③ 查看 test topic 的详情
[rootzk-kfk01 config]# kafka-topics.sh --describe --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181 test
Topic: test PartitionCount: 3 ReplicationFactor: 2 Configs:
# 表示主题 test 共有 3 个分区。每个分区的复制因子为 2即每个分区都有两个副本Topic: test Partition: 0 Leader: 0 Replicas: 0,2 Isr: 0,2Topic: test Partition: 1 Leader: 1 Replicas: 1,0 Isr: 1,0Topic: test Partition: 2 Leader: 2 Replicas: 2,1 Isr: 2,1
# Partition: x 指明了每个分区的编号。
# Leader: x 显示了每个分区的 leader 副本所在的 broker 编号。
# Replicas: x,y 展示了每个分区的所有副本所在的 broker 编号。
# Isr: x,y 描述了每个分区的“在同步副本集”In-Sync Replica中的副本编号④ 发布消息
[rootzk-kfk01 config]# kafka-console-producer.sh --broker-list 192.168.190.104:9092,192.168.190.105:9092,192.168.190.106:9092 --topic test
123456⑤ 消费消息
[rootzk-kfk02 config]# kafka-console-consumer.sh --bootstrap-server 192.168.190.104:9092,192.168.190.105:9092,192.168.190.106:9092 --topic test --from-beginning
123456[rootzk-kfk03 config]# kafka-console-consumer.sh --bootstrap-server 192.168.190.104:9092,192.168.190.105:9092,192.168.190.106:9092 --topic test --from-beginning
123456
# --from-beginning会把主题中以往所有的数据都读取出来
⑥ 修改分区数
修改主题 test 的分区数量为 6
kafka-topics.sh --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181 --alter --topic test --partitions 6
⑦ 删除 topic
kafka-topics.sh --delete --zookeeper 192.168.190.104:2181,192.168.190.105.21:2181,192.168.190.106:2181 --topic test
7. 创建主题一般故障
故障示例
[rootzk-kfk01 config]# kafka-topics.sh --create --zookeeper 192.168.190.104:2181,192.168.190.105:2181,192.168.190.106:2181 --replication-factor 2 --partitions 3 --topic test
Error while executing topic command : Replication factor: 2 larger than available brokers: 0.
[2024-04-13 20:17:55,154] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 2 larger than available brokers: 0.
# 指定的副本因子大于可用的 broker 数量。这可能是由于 ZooKeeper 中注册的 broker 数量与实际运行的 broker 数量不匹配导致的。
排查过程
① 确保 broker 数量足够
② 查看 server.properties 配置
broker.id
listenersPLAINTEXT://ip:9092
确保编号唯一监听地址是否正确
③ 查看端口是否正常
telenet 主机名 2181
④ 关闭 kafka再挨个启动
service kafka stop
service kafka start
五、KafkaELKF
ELKF 部署请参考ELK、ELKF企业级日志分析系统介绍-CSDN博客
环境配置 服务器ip 节点名称myid/角色软件版本jdk版本操作系统192.168.190.100node1elasticsearch-5.5.0、kibana-5.5.1openjdk version 1.8.0_131centos7.4192.168.190.101node2elasticsearch-5.5.0openjdk version 1.8.0_131centos7.4192.168.190.102logstashapache、logstash-5.5.1centos7.4192.168.190.103filebeatfilebeat、apachecentos7.4192.168.190.104zk-kfk011/followzookeeper-3.5.7openjdk version 1.8.0_131centos7.4192.168.190.105zk-kfk022/followzookeeper-3.5.7openjdk version 1.8.0_131centos7.4192.168.190.106zk-kfk033/leaderzookeeper-3.5.7openjdk version 1.8.0_131centos7.4
数据流向 1. 修改 filebeat 的主配置文件
[rootfilebeat ~]# cd /usr/local/filebeat
[rootfilebeat filebeat]# vim filebeat.yml
filebeat.prospectors:
- type: logenabled: truepaths:- /var/log/httpd/access_logtags: [access]- type: logenabled: truepaths:- /var/log/httpd/error_logtags: [error]
……
注释 Logstash output 相关配置避免冲突
#----------------------------- kafka output --------------------------------
output.kafka:enabled: truehosts: [192.168.190.104:9092,192.168.190.105:9092,192.168.190.106:9092]
# 指定 Kafka 集群配置topic: httpd # 指定 Kafka 的 topic
[rootfilebeat filebeat]# vim filebeat.yml # 启动 filebeat
2. 在 filebeat 节点安装 apache
[rootfilebeat filebeat]# yum install -y httpd
[rootfilebeat filebeat]# systemctl start httpd.service
3. 新建一个 Logstash kafka 配置文件
[rootlogstash ~]# cd /etc/logstash/conf.d/
[rootlogstash conf.d]# vim kafka.conf
input {kafka {bootstrap_servers 192.168.190.104:9092,192.168.190.105:9092,192.168.190.106:9092
# kafka集群地址topics httpd # 拉取的kafka的指定topictype httpd_kafka # 指定 type 字段codec json # 解析json格式的日志数据auto_offset_reset latest # 拉取最近数据earliest为从头开始拉取decorate_events true # 传递给elasticsearch的数据额外增加kafka的属性数据}
}output {if access in [tags] {elasticsearch {hosts [192.168.190.100:9200]index httpd_access-%{YYYY.MM.dd}}}if error in [tags] {elasticsearch {hosts [192.168.190.100:9200]index httpd_error-%{YYYY.MM.dd}}}stdout { codec rubydebug }
}
[rootlogstash conf.d]# logstash -f kafka.conf
# 启动 logstash或者 ./filebeat -e -c filebeat.yml 网页端访问apache
http://192.168.190.102/
4. 查看所有的索引
生产黑屏操作es时查看所有的索引
[rootnode1 ~]# curl -X GET localhost:9200/_cat/indices?v
health status index uuid pri rep docs.count docs.deleted store.size pri.store.size
green open httpd_error-2024.04.13 zHoWjUNPR2uggajGdK13_g 5 1 36 0 716.6kb 358.3kb
green open system-2024.04.10 A8E0O-f9Q4OsN5SsoqkXDw 5 1 7132 0 4.4mb 2.2mb
green open index-demo1 3voNtLzAQIqDJn3Ip2ODKA 5 1 1 0 9.3kb 4.6kb
green open logstash-2024.04.10 1KY7jcRlSfGHhtklVpIchg 5 1 2 0 18kb 9kb
green open httpd_access-2024.04.13 0EK1CjoZRCmOboLgtVhesQ 5 1 60 0 260.3kb 113.9kb
green open apache_error-2024.04.10 un_LRAJzTSmL9aq6DUL5zg 5 1 13 0 124.5kb 62.2kb
green open filebeat-2024.04.11 rX5Vy-73Q1aSUKBI3LGe0Q 5 1 6031 0 3.9mb 1.9mb
green open .kibana ev0JtIH6SyOpycSjI2TGQQ 1 1 6 1 69.7kb 34.8kb
green open apache_access-2024.04.10 lCL7_nw3QDmGr2YmbYUOuQ 5 1 12 0 113.8kb 56.9kbes-head web 界面查看 5. 登录 Kibana 添加索引
浏览器访问 http://192.168.190.100:5601添加索引“httpd*”查看图表信息及日志信息。
