网站 动态内容加速,企业网站推广哪些效果比较好,室内设计联盟免费下载,旅游o2o小程序源码文章目录 KafkaKafka 概述使用消息队列的好处 Kafka 的特性Kafka 系统架构Kafka 的应用场景Kafka 的优缺点 Kafka 集群部署下载安装包安装 KafkaKafka 命令行操作Kafka 架构深入 FilebeatKafkaELK 部署指南~部署 ZookeeperKafka 集群部署 Filebeat部署 ELK#xff08;Logstash… 文章目录 KafkaKafka 概述使用消息队列的好处 Kafka 的特性Kafka 系统架构Kafka 的应用场景Kafka 的优缺点 Kafka 集群部署下载安装包安装 KafkaKafka 命令行操作Kafka 架构深入 FilebeatKafkaELK 部署指南~部署 ZookeeperKafka 集群部署 Filebeat部署 ELKLogstash 配置Kibana 配置与查看日志 Kafka
Kafka 概述
Kafka 是一个分布式、基于发布/订阅模式的消息队列系统由 Linkedin 开发并贡献给 Apache 基金会现已成为顶级开源项目。它主要应用于大数据领域的实时计算以及日志收集具有高吞吐量、低延迟、可扩展性、持久性、可靠性、容错性和高并发的特性。
使用消息队列的好处
解耦允许独立地扩展或修改两边的处理过程只要确保它们遵守同样的接口约束。可恢复性系统的一部分组件失效时不会影响到整个系统。缓冲有助于控制和优化数据流经过系统的速度解决生产消息和消费消息的处理速度不一致的情况。灵活性 峰值处理能力使用消息队列能够使关键组件顶住突发的访问压力。异步通信允许用户把一个消息放入队列但并不立即处理它。
Kafka 的特性
高吞吐量、低延迟每秒可以处理几十万条消息延迟最低只有几毫秒。可扩展性Kafka 集群支持热扩展。持久性、可靠性消息被持久化到本地磁盘并且支持数据备份防止数据丢失。容错性允许集群中节点失败。高并发支持数千个客户端同时读写。
Kafka 系统架构
Broker 一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic。 Topic 可以理解为一个队列生产者和消费者面向的都是一个 topic。类似于数据库的表名或者 ES 的 index。物理上不同 topic 的消息分开存储。 Partition 为了实现扩展性一个非常大的 topic 可以分布到多个 broker 上。一个 topic 可以分割为一个或多个 partition每个 partition 是一个有序的队列。Kafka 只保证 partition 内的记录是有序的。数据路由规则指定了 partition 则直接使用未指定但指定 key通过对 key 的 value 进行 hash 取模选出一个 partition都未指定使用轮询选出一个 partition。每个 partition 中的数据使用多个 segment 文件存储。 Replica 副本机制为保证集群中的某个节点发生故障时该节点上的 partition 数据不丢失且 Kafka 仍然能够继续工作。一个 topic 的每个分区都有若干个副本一个 leader 和若干个 follower。 Leader 当前负责数据的读写的 partition。 Follower 跟随 Leader所有写请求都通过 Leader 路由。数据变更会广播给所有 Follower与 Leader 保持数据同步。只负责备份不负责数据的读写。如果 Leader 故障则从 Follower 中选举出一个新的 Leader。 Producer 数据的发布者将消息 push 发布到 Kafka 的 topic 中。 Consumer 从 broker 中 pull 拉取数据。可以消费多个 topic 中的数据。 Consumer GroupCG 由多个 consumer 组成。所有的消费者都属于某个消费者组。消费者组内每个消费者负责消费不同分区的数据防止数据被重复读取。消费者组之间互不影响。 Offset 偏移量 唯一标识一条消息。决定读取数据的位置。消费者通过偏移量来决定下次读取的消息。消息被消费之后并不被马上删除。某一个业务也可以通过修改偏移量达到重新读取消息的目的。消息默认生命周期为 1 周7*24小时。 Zookeeper 在 Kafka 中ZooKeeper 负责维护 Kafka 集群的一些元数据和 leader 选举等协调工作。元数据存储存储主题、分区、Broker 节点等信息。Leader 选举参与领导者选举的过程。健康监控进行集群的健康监控。消费者组协调协调和追踪消费者的位置信息。
Kafka 的应用场景
日志收集Kafka 可以被用作日志收集系统将各种应用的日志数据集中收集起来方便后续的处理和分析。实时计算Kafka 可以作为实时计算系统的数据源如 Spark Streaming、Flink 等用于实时数据处理和分析。消息通讯Kafka 可以作为消息通讯系统实现不同系统之间的数据交换和通信。流量削峰在高并发场景下Kafka 可以作为流量削峰的工具将大量的请求缓存到 Kafka 中然后按照一定的速率进行处理避免系统崩溃。
Kafka 的优缺点
优点
高吞吐量、低延迟。可扩展性强。持久性、可靠性高。支持多副本、容错性强。社区活跃、生态丰富。
缺点
依赖 Zookeeper如果 Zookeeper 出现故障会影响 Kafka 的正常运行。数据一致性方面虽然 Kafka 提供了多副本机制但是在极端情况下仍然可能存在数据丢失的风险。消息顺序问题如果生产者发送消息到多个分区那么消费者消费时可能无法保证消息的顺序性。
Kafka 集群部署
下载安装包
官方下载地址Apache Kafka 下载页面步骤 切换到 /opt 目录。使用 wget 从清华大学镜像站下载 Kafka 2.7.1 版本。
cd /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz安装 Kafka
步骤 解压 Kafka 压缩包。将解压后的目录移动到 /usr/local/kafka。备份并编辑 server.properties 文件配置 Kafka。
cd /opt/
tar zxvf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafkacd /usr/local/kafka/config/
cp server.properties{,.bak}
vim server.properties关键配置项 broker.id每个 Kafka 实例的唯一标识集群中每个实例的 broker.id 必须不同。listeners指定 Kafka 监听的 IP 和端口。num.network.threads 和 num.io.threads分别设置处理网络请求和磁盘 IO 的线程数。log.dirsKafka 数据和日志的存放路径。zookeeper.connect指定 Zookeeper 集群的地址。 环境变量配置 将 Kafka 的 bin 目录添加到 PATH 环境变量中。
vim /etc/profile
export KAFKA_HOME/usr/local/kafka
export PATH$PATH:$KAFKA_HOME/bin
source /etc/profile配置启动脚本 创建一个 Kafka 的启动脚本并设置开机自启。
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME/usr/local/kafka
case $1 in
start)echo ---------- Kafka 启动 ------------${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)echo ---------- Kafka 停止 ------------${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)$0 stop$0 start
;;
status)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}
esacchmod x /etc/init.d/kafka
chkconfig --add kafka
service kafka startKafka 命令行操作
创建 topic
kafka-topics.sh --create --zookeeper 192.168.80.10:2181,192.168.80.11:2181,192.168.80.12:2181 --replication-factor 2 --partitions 3 --topic test查看 topic
kafka-topics.sh --list --zookeeper 192.168.80.10:2181,192.168.80.11:2181,192.168.80.12:2181
kafka-topics.sh --describe --zookeeper 192.168.80.10:2181,192.168.80.11:2181,192.168.80.12:2181 --topic test发布和消费消息
# 生产者
kafka-console-producer.sh --broker-list 192.168.80.10:9092,192.168.80.11:9092,192.168.80.12:9092 --topic test# 消费者
kafka-console-consumer.sh --bootstrap-server 192.168.80.10:9092,192.168.80.11:9092,192.168.80.12:9092 --topic test --from-beginning修改和删除 topic
# 修改分区数
kafka-topics.sh --zookeeper 192.168.80.10:2181,192.168.80.11:2181,192.168.80.12:2181 --alter --topic test --partitions 6# 删除 topic
kafka-topics.sh --delete --zookeeper 192.168.80.10:2181,192.168.80.11:2181,192.168.80.12:2181 --topic testKafka 架构深入 工作流程及文件存储机制 Kafka 以 topic 对消息进行分类producer 和 consumer 都是面向 topic 的。Topic 是逻辑概念partition 是物理概念每个 partition 对应一个 log 文件。为防止 log 文件过大Kafka 采用分片和索引机制将每个 partition 分为多个 segment每个 segment 包含 .index 和 .log 文件。 数据可靠性保证 Kafka 通过 ack 应答机制保证数据可靠性producer 发送数据后需要等待 broker 的确认。 数据一致性问题 LEO每个副本的最大 offset。HW消费者能见到的最大 offset所有副本中最小的 LEO。Leader 和 follower 故障时的数据恢复和同步机制。 ack 应答机制 Kafka 提供了三种可靠性级别acks0, 1, -1用户可以根据需求选择。幂等性在 0.11 版本及以后Kafka 引入了幂等性特性保证 producer 发送重复数据时server 端只持久化一条。
注释
Kafka 的安装和配置需要根据集群的实际环境进行调整特别是 IP 地址和端口号。在生产环境中通常需要配置更多的参数以优化性能和可靠性。Kafka 的数据可靠性和一致性机制是其核心特性之一理解这些机制对于保证数据的安全性和一致性至关重要。
FilebeatKafkaELK 部署指南~
部署 ZookeeperKafka 集群
目的搭建消息队列系统用于日志数据的传输。步骤 安装并配置 Zookeeper 集群。安装并配置 Kafka 集群指定 Zookeeper 集群地址。启动 Zookeeper 和 Kafka 服务确保集群正常运行。
部署 Filebeat
目的收集服务器上的日志数据。步骤 下载并解压 Filebeat 到指定目录如 /usr/local/filebeat。编辑 filebeat.yml 配置文件filebeat.prospectors:
- type: logenabled: truepaths:- /var/log/httpd/access_logtags: [access]- type: logenabled: truepaths:- /var/log/httpd/error_logtags: [error]# 添加输出到 Kafka 的配置
output.kafka:enabled: truehosts: [192.168.80.10:9092,192.168.80.11:9092,192.168.80.12:9092]topic: httpd启动 Filebeat开始收集日志并发送到 Kafka。
部署 ELKLogstash 配置
目的从 Kafka 拉取日志数据并处理、存储到 Elasticsearch 中。步骤 在 Logstash 组件所在节点上新建一个 Logstash 配置文件 kafka.confinput {kafka {bootstrap_servers 192.168.80.10:9092,192.168.80.11:9092,192.168.80.12:9092topics httpdtype httpd_kafkacodec jsonauto_offset_reset latestdecorate_events true}
}output {if access in [tags] {elasticsearch {hosts [192.168.80.30:9200]index httpd_access-%{YYYY.MM.dd}}}if error in [tags] {elasticsearch {hosts [192.168.80.30:9200]index httpd_error-%{YYYY.MM.dd}}}stdout { codec rubydebug }
}启动 Logstash开始从 Kafka 拉取日志并存储到 Elasticsearch。
Kibana 配置与查看日志
目的通过 Kibana 可视化界面查看日志数据。步骤 在浏览器中访问 Kibana如 http://192.168.80.30:5601。登录 Kibana如果设置了登录认证。单击“Create Index Pattern”按钮添加索引模式例如 httpd_access-* 和 httpd_error-*注意这里应与 Logstash 配置中的 index 名称匹配但原笔记中的 filebeat_test-* 是不正确的。单击“create”按钮创建索引模式。单击“Discover”按钮可查看图表信息及日志信息。
注释
在配置 Filebeat 和 Logstash 时确保 Kafka 集群的地址和 topic 名称正确无误。Logstash 的 auto_offset_reset 参数决定了从 Kafka 拉取数据的起始位置latest 表示从最新的数据开始拉取earliest 表示从头开始拉取。Kibana 中的索引模式应与 Logstash 配置中的 index 名称一致以便正确显示日志数据。在实际部署中还需要考虑安全性、性能优化等方面的问题。
