【企业级分布式系统】 Kafka集群

文章目录

  • Kafka
    • Kafka 概述
      • 使用消息队列的好处
    • Kafka 的特性
    • Kafka 系统架构
      • Kafka 的应用场景
      • Kafka 的优缺点
  • Kafka 集群部署
    • 下载安装包
    • 安装 Kafka
    • Kafka 命令行操作
    • Kafka 架构深入
  • Filebeat+Kafka+ELK 部署指南~
    • 部署 Zookeeper+Kafka 集群
    • 部署 Filebeat
    • 部署 ELK(Logstash 配置)
    • Kibana 配置与查看日志

Kafka

Kafka 概述

Kafka 是一个分布式、基于发布/订阅模式的消息队列系统,由 Linkedin 开发并贡献给 Apache 基金会,现已成为顶级开源项目。它主要应用于大数据领域的实时计算以及日志收集,具有高吞吐量、低延迟、可扩展性、持久性、可靠性、容错性和高并发的特性。

使用消息队列的好处

  1. 解耦:允许独立地扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
  2. 可恢复性:系统的一部分组件失效时,不会影响到整个系统。
  3. 缓冲:有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。
  4. 灵活性 & 峰值处理能力:使用消息队列能够使关键组件顶住突发的访问压力。
  5. 异步通信:允许用户把一个消息放入队列,但并不立即处理它。

Kafka 的特性

  • 高吞吐量、低延迟:每秒可以处理几十万条消息,延迟最低只有几毫秒。
  • 可扩展性:Kafka 集群支持热扩展。
  • 持久性、可靠性:消息被持久化到本地磁盘,并且支持数据备份防止数据丢失。
  • 容错性:允许集群中节点失败。
  • 高并发:支持数千个客户端同时读写。

Kafka 系统架构

  1. Broker
    • 一台 Kafka 服务器就是一个 broker。
    • 一个集群由多个 broker 组成。
    • 一个 broker 可以容纳多个 topic。
  2. Topic
    • 可以理解为一个队列,生产者和消费者面向的都是一个 topic。
    • 类似于数据库的表名或者 ES 的 index。
    • 物理上不同 topic 的消息分开存储。
  3. Partition
    • 为了实现扩展性,一个非常大的 topic 可以分布到多个 broker 上。
    • 一个 topic 可以分割为一个或多个 partition,每个 partition 是一个有序的队列。
    • Kafka 只保证 partition 内的记录是有序的。
    • 数据路由规则:指定了 partition 则直接使用;未指定但指定 key,通过对 key 的 value 进行 hash 取模选出一个 partition;都未指定,使用轮询选出一个 partition。
    • 每个 partition 中的数据使用多个 segment 文件存储。
  4. Replica
    • 副本机制,为保证集群中的某个节点发生故障时,该节点上的 partition 数据不丢失,且 Kafka 仍然能够继续工作。
    • 一个 topic 的每个分区都有若干个副本,一个 leader 和若干个 follower。
  5. Leader
    • 当前负责数据的读写的 partition。
  6. Follower
    • 跟随 Leader,所有写请求都通过 Leader 路由。
    • 数据变更会广播给所有 Follower,与 Leader 保持数据同步。
    • 只负责备份,不负责数据的读写。
    • 如果 Leader 故障,则从 Follower 中选举出一个新的 Leader。
  7. Producer
    • 数据的发布者,将消息 push 发布到 Kafka 的 topic 中。
  8. Consumer
    • 从 broker 中 pull 拉取数据。
    • 可以消费多个 topic 中的数据。
  9. Consumer Group(CG)
    • 由多个 consumer 组成。
    • 所有的消费者都属于某个消费者组。
    • 消费者组内每个消费者负责消费不同分区的数据,防止数据被重复读取。
    • 消费者组之间互不影响。
  10. Offset 偏移量
    • 唯一标识一条消息。
    • 决定读取数据的位置。
    • 消费者通过偏移量来决定下次读取的消息。
    • 消息被消费之后,并不被马上删除。
    • 某一个业务也可以通过修改偏移量达到重新读取消息的目的。
    • 消息默认生命周期为 1 周(7*24小时)。
  11. Zookeeper
    • 在 Kafka 中,ZooKeeper 负责维护 Kafka 集群的一些元数据和 leader 选举等协调工作。
    • 元数据存储:存储主题、分区、Broker 节点等信息。
    • Leader 选举:参与领导者选举的过程。
    • 健康监控:进行集群的健康监控。
    • 消费者组协调:协调和追踪消费者的位置信息。

Kafka 的应用场景

  1. 日志收集:Kafka 可以被用作日志收集系统,将各种应用的日志数据集中收集起来,方便后续的处理和分析。
  2. 实时计算:Kafka 可以作为实时计算系统的数据源,如 Spark Streaming、Flink 等,用于实时数据处理和分析。
  3. 消息通讯:Kafka 可以作为消息通讯系统,实现不同系统之间的数据交换和通信。
  4. 流量削峰:在高并发场景下,Kafka 可以作为流量削峰的工具,将大量的请求缓存到 Kafka 中,然后按照一定的速率进行处理,避免系统崩溃。

Kafka 的优缺点

优点

  • 高吞吐量、低延迟。
  • 可扩展性强。
  • 持久性、可靠性高。
  • 支持多副本、容错性强。
  • 社区活跃、生态丰富。

缺点

  • 依赖 Zookeeper,如果 Zookeeper 出现故障,会影响 Kafka 的正常运行。
  • 数据一致性方面,虽然 Kafka 提供了多副本机制,但是在极端情况下,仍然可能存在数据丢失的风险。
  • 消息顺序问题,如果生产者发送消息到多个分区,那么消费者消费时可能无法保证消息的顺序性。

Kafka 集群部署

下载安装包

  • 官方下载地址:Apache Kafka 下载页面
  • 步骤
    1. 切换到 /opt 目录。
    2. 使用 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

  • 步骤
    1. 解压 Kafka 压缩包。
    2. 将解压后的目录移动到 /usr/local/kafka
    3. 备份并编辑 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.threadsnum.io.threads:分别设置处理网络请求和磁盘 IO 的线程数。
    • log.dirs:Kafka 数据和日志的存放路径。
    • 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 running"elseecho "kafka is running"fi
;;
*)echo "Usage: $0 {start|stop|restart|status}"
esacchmod +x /etc/init.d/kafka
chkconfig --add kafka
service kafka start

Kafka 命令行操作

  • 创建 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 test

Kafka 架构深入

  • 工作流程及文件存储机制

    • 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 提供了三种可靠性级别(acks=0, 1, -1),用户可以根据需求选择。
    • 幂等性:在 0.11 版本及以后,Kafka 引入了幂等性特性,保证 producer 发送重复数据时,server 端只持久化一条。

注释

  • Kafka 的安装和配置需要根据集群的实际环境进行调整,特别是 IP 地址和端口号。
  • 在生产环境中,通常需要配置更多的参数以优化性能和可靠性。
  • Kafka 的数据可靠性和一致性机制是其核心特性之一,理解这些机制对于保证数据的安全性和一致性至关重要。

Filebeat+Kafka+ELK 部署指南~

部署 Zookeeper+Kafka 集群

  • 目的:搭建消息队列系统,用于日志数据的传输。
  • 步骤
    1. 安装并配置 Zookeeper 集群。
    2. 安装并配置 Kafka 集群,指定 Zookeeper 集群地址。
    3. 启动 Zookeeper 和 Kafka 服务,确保集群正常运行。

部署 Filebeat

  • 目的:收集服务器上的日志数据。
  • 步骤
    1. 下载并解压 Filebeat 到指定目录(如 /usr/local/filebeat)。
    2. 编辑 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"
      
    3. 启动 Filebeat,开始收集日志并发送到 Kafka。

部署 ELK(Logstash 配置)

  • 目的:从 Kafka 拉取日志数据,并处理、存储到 Elasticsearch 中。
  • 步骤
    1. 在 Logstash 组件所在节点上,新建一个 Logstash 配置文件 kafka.conf
      input {kafka {bootstrap_servers => "192.168.80.10:9092,192.168.80.11:9092,192.168.80.12:9092"topics  => "httpd"type => "httpd_kafka"codec => "json"auto_offset_reset => "latest"decorate_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 }
      }
      
    2. 启动 Logstash,开始从 Kafka 拉取日志并存储到 Elasticsearch。

Kibana 配置与查看日志

  • 目的:通过 Kibana 可视化界面查看日志数据。
  • 步骤
    1. 在浏览器中访问 Kibana(如 http://192.168.80.30:5601)。
    2. 登录 Kibana(如果设置了登录认证)。
    3. 单击“Create Index Pattern”按钮,添加索引模式,例如 httpd_access-*httpd_error-*(注意:这里应与 Logstash 配置中的 index 名称匹配,但原笔记中的 filebeat_test-* 是不正确的)。
    4. 单击“create”按钮创建索引模式。
    5. 单击“Discover”按钮,可查看图表信息及日志信息。

注释

  • 在配置 Filebeat 和 Logstash 时,确保 Kafka 集群的地址和 topic 名称正确无误。
  • Logstash 的 auto_offset_reset 参数决定了从 Kafka 拉取数据的起始位置,latest 表示从最新的数据开始拉取,earliest 表示从头开始拉取。
  • Kibana 中的索引模式应与 Logstash 配置中的 index 名称一致,以便正确显示日志数据。
  • 在实际部署中,还需要考虑安全性、性能优化等方面的问题。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/61154.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Java-空链处理

什么是 null 在 Java 中,null 是一个非常常见的关键字,用于表示“没有值”或“空”。然而,对于初学者来说,null 的本质可能会感到有些困惑。在本文中,我们将详细探讨 null 在 Java 中的含义和使用。 在 Java 中&…

PyTorch使用教程-深度学习框架

PyTorch使用教程-深度学习框架 1. PyTorch简介 1.1-什么是PyTorch ​ PyTorch是一个广泛使用的开源机器学习框架,特别适合深度学习的应用。它以其动态计算图而闻名,允许在运行时修改模型,使得实验和调试更加灵活。PyTorch提供了强大的GPU加…

供应链管理、一件代发系统功能及源码分享 PHP+Mysql

随着电商行业的不断发展,传统的库存管理模式已经逐渐无法满足市场需求。越来越多的企业选择“一件代发”模式,即商家不需要自己储备商品库存,而是将订单直接转给供应商,由供应商直接进行发货。这种方式极大地降低了企业的运营成本…

Pr:音频过渡

Adobe Premiere Pro 自带一组共三个音频过渡 Audio Transitions效果。 对音频剪辑之间应用交叉淡化 Crossfade过渡,操作方式类似于应用视频过渡效果。 对于交叉淡化,要保证前剪辑的出点之后及后剪辑的入点之前有足够的预留内容(也称“手柄”&…

前端页面开发步骤详解

目录 前言1. 页面搭建1.1 HTML 标签结构1.2 CSS 样式设计 2. 数据绑定与事件处理2.1 数据绑定2.2 表单校验 3. 调用后台接口3.1 接口文档与工具封装3.2 参数传递与接口调用 结语 前言 在前端开发过程中,从页面搭建到与后台接口对接是一个必不可少的完整流程。无论是…

A037-基于Spring Boot的二手物品交易的设计与实现

🙊作者简介:在校研究生,拥有计算机专业的研究生开发团队,分享技术代码帮助学生学习,独立完成自己的网站项目。 代码可以查看文章末尾⬇️联系方式获取,记得注明来意哦~🌹 赠送计算机毕业设计600…

两行命令搭建深度学习环境(Docker/torch2.5.1+cu118/命令行美化+插件),含完整的 Docker 安装步骤

深度学习环境的配置过于繁琐,所以我制作了两个基础的镜像,希望可以帮助大家节省时间,你可以选择其中一种进行安装,版本说明: base 版本基于 pytorch/pytorch:2.5.1-cuda11.8-cudnn9-devel,默认 python 版本…

EXCEL延迟退休公式

如图: A B为手工输入 C2EOMONTH(A2,B2*12) D2EOMONTH(C2,IF(C2>DATEVALUE("2025-1-1"),INT((DATEDIF(DATEVALUE("2025-1-1"),C2,"m")4)/4),0)) E2EOMONTH(A2,B2*12IF(EOMONTH(A2,B2*12)>DATEVALUE("2025-1-1"),INT(…

区块链技术在数据安全中的应用

💓 博客主页:瑕疵的CSDN主页 📝 Gitee主页:瑕疵的gitee主页 ⏩ 文章专栏:《热点资讯》 区块链技术在数据安全中的应用 区块链技术在数据安全中的应用 区块链技术在数据安全中的应用 引言 区块链技术基础 1.1 区块链的…

GIT 入门详解指南

前言: 注:本博客仅用于记录本人学习过程中对git的理解,仅供学习参考,如有异议请自行查资料求证 安装 使用git之前必须完成git的安装,Git 目前支持 Linux/Unix、Solaris、Mac和 Windows 平台上运行 git 安装教程 基本…

讯飞、阿里云、腾讯云:Android 语音合成服务对比选择

在 移动端 接入语音合成方面,讯飞和腾讯云等都是优秀的选择,但各有其特点和优势。咱们的需求是需要支持普通话/英语/法语三种语言,以下是对各个平台的详细比较: 一、讯飞语音合成介绍 与语音听写相反,语音合成是将一段…

HarmonyOS本地存储-Preferences(用户首选项)的使用

一,用户首选项简述 ohos.data.preferences (用户首选项) 用户首选项为应用提供Key-Value键值型的数据处理能力,支持应用持久化轻量级数据,并对其修改和查询。 数据存储形式为键值对,键的类型为字符串型,值的存储数据…

【机器学习】回归模型(线性回归+逻辑回归)原理详解

线性回归 Linear Regression 1 概述 线性回归类似高中的线性规划题目。线性回归要做的是就是找到一个数学公式能相对较完美地把所有自变量组合(加减乘除)起来,得到的结果和目标接近。 线性回归分为一元线性回归和多元线性回归。 2 一元线…

OceanBase 分区表详解

1、分区表的定义 在OceanBase数据库中,普通的表数据可以根据预设的规则被分割并存储到不同的数据区块中,同一区块的数据是在一个物理存储上。这样被分区块的表被称为分区表,而其中的每一个独立的数据区块则被称为一个分区。 如下图所示&…

【Android原生问题分析】夸克、抖音划动无响应问题【Android14】

1 问题描述 偶现问题,用户打开夸克、抖音后,在界面上划动无响应,但是没有ANR。回到Launcher后再次打开夸克/抖音,发现App的界面发生了变化,但是仍然是划不动的。 2 log初分析 复现问题附近的log为: 用户…

使用 K-means 算法进行豆瓣读书数据的文本聚类分析

✅作者简介:2022年博客新星 第八。热爱国学的Java后端开发者,修心和技术同步精进。 🍎个人主页:Java Fans的博客 🍊个人信条:不迁怒,不贰过。小知识,大智慧。 💞当前专栏…

Django5 2024全栈开发指南(二):Django项目配置详解

目录 一、基本配置信息二、资源文件配置2.1 资源路由——STATIC_URL2.2 资源集合——STATICFILES_DIRS2.3 资源部署——STATIC_ROOT2.2.4 媒体资源——MEDIA 三、模板配置四、数据库配置4.1 mysqlclient连接MySQL4.2 pymysql连接MySQL4.3 多个数据库的连接方式4.4 使用配置文件…

数据结构-二叉搜索树(Java语言)

目录 1.概念 2.查找search 3.插入insert ​编辑4.删除remove(难点) 5.性能分析 1.概念 二叉搜索树又称二叉排序树,它或者是一棵空树,或者是具有以下性质的二叉树 : 1.若它的左子树不为空,则左子树上所有节点的值都…

学习笔记:黑马程序员JavaWeb开发教程(2024.11.18)

9.8 Mybatis-基础操作-查询(条件查询) 需要模糊查询,根据要求,我们需要在关键词前后都加上%,但是我们不能使用‘%#{内容}%’的形式,因为#{内容}最终会变成?,而?不能放在‘’之中&#xff…

数据分析-48-时间序列变点检测之在线实时数据的CPD

文章目录 1 时间序列结构1.1 变化点的定义1.2 结构变化的类型1.2.1 水平变化1.2.2 方差变化1.3 变点检测1.3.1 离线数据检测方法1.3.2 实时数据检测方法2 模拟数据2.1 模拟恒定方差数据2.2 模拟变化方差数据3 实时数据CPD3.1 SDAR学习算法3.2 Changefinder模块3.3 恒定方差CPD3…