AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Kafka入门宝典（详细截图版）（一）

2022-05-26 217
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云原生网关 MSE Higress，422元/月
注册配置 MSE Nacos/ZooKeeper，118元/月
日志服务 SLS，月写入数据量 50GB 1个月
简介： Kafka入门宝典（详细截图版）（一）

1、了解 Apache Kafka

1.1、简介

image.png

官网：http://kafka.apache.org/

  • Apache Kafka 是一个开源消息系统，由Scala 写成。是由Apache 软件基金会开发的一个开源消息系统项目。
  • Kafka 最初是由LinkedIn 开发，并于2011 年初开源。2012 年10 月从Apache Incubator 毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待（低延时）的平台。
  • Kafka 是一个分布式消息系统：具有生产者、消费者的功能。它提供了类似于JMS 的特性，但是在设计实现上完全不同，此外它并不是JMS 规范的实现。【重点】

1.2、kafka的基本结构

image.png

  • Producer：消息的发送者
  • Consumer：消息的接收者
  • kafka cluster：kafka的集群。
  • Topic：就是消息类别名，一个topic中通常放置一类消息。每个topic都有一个或者多个订阅者（消费者）。

消息的生产者将消息推送到kafka集群，消息的消费者从kafka集群中拉取消息。

1.3、kafka的完整架构

image.png

说明：

  • broker：集群中的每一个kafka实例，称之为broker；
  • ZooKeeper：Kafka 利用ZooKeeper 保存相应元数据信息， Kafka 元数据信息包括如代理节点信息、Kafka集群信息、旧版消费者信息及其消费偏移量信息、主题信息、分区状态信息、分区副本分配方案信息、动态配置信息等。
  • ConsumerGroup：在Kafka 中每一个消费者都属于一个特定消费组（ ConsumerGroup ），我们可以为每个消费者指定一个消费组，以groupld 代表消费组名称，通过group.id 配置设置。如果不指定消费组，则该消费者属于默认消费组test-consumer-group 。

1.4、kafka的特性

  • 消息持久化
  • Kafka 基于文件系统来存储和缓存消息。
  • 高吞吐量
  • Kafka 将数据写到磁盘，充分利用磁盘的顺序读写。同时， Kafka 在数据写入及数据同步采用了零拷贝（ zero-copy ）技术，采用sendFile()函数调用，sendFile()函数是在两个文件描述符之间直接传递数据，完全在内核中操作，从而避免了内核缓冲区与用户缓冲区之间数据的拷贝，操作效率极高。
  • Kafka 还支持数据压缩及批量发送，同时Kafka 将每个主题划分为多个分区，这一系列的优化及实现方法使得Kafka 具有很高的吞吐量。经大多数公司对Kafka 应用的验证， Kafka 支持每秒数百万级别的消息。
  • 高扩展性
  • Kafka 依赖ZooKeeper来对集群进行协调管理，这样使得Kafka 更加容易进行水平扩展，生产者、消费者和代理都为分布式，可配置多个。
  • 同时在机器扩展时无需将整个集群停机，集群能够自动感知，重新进行负责均衡及数据复制。
  • 多客户端支持
  • Kafka 核心模块用Scala 语言开发，Kafka 提供了多种开发语言的接入，如Java 、Scala、C 、C＋＋、Python 、Go 、Erlang 、Ruby 、Node. 等。
  • 安全机制
  • 通过SSL 和SASL(Kerberos), SASL/PLA时验证机制支持生产者、消费者与broker连接时的身份认证；
  • 支持代理与ZooKeeper 连接身份验证；
  • 通信时数据加密；
  • 客户端读、写权限认证；
  • Kafka 支持与外部其他认证授权服务的集成；
  • Kafka 支持以下几种安全措施：
  • 数据备份
  • Kafka 可以为每个topic指定副本数，对数据进行持久化备份，这可以一定程度上防止数据丢失，提高可用性。
  • 轻量级
  • Kafka 的实例是无状态的，即broker不记录消息是否被消费，消费偏移量的管理交由消费者自己或组协调器来维护。
  • 同时集群本身几乎不需要生产者和消费者的状态信息，这就使得Kafka非常轻量级，同时生产者和消费者客户端实现也非常轻量级。
  • 消息压缩
  • Kafka 支持Gzip, Snappy 、LZ4 这3 种压缩方式，通常把多条消息放在一起组成MessageSet，然后再把Message Set 放到一条消息里面去，从而提高压缩比率进而提高吞吐量。

1.5、kafka的应用场景

  • 消息系统。
  • Kafka 作为一款优秀的消息系统，具有高吞吐量、内置的分区、备份冗余分布式等特点，为大规模消息处理提供了一种很好的解决方案。
  • 应用监控。
  • 利用Kafka 采集应用程序和服务器健康相关的指标，如CPU 占用率、IO 、内存、连接数、TPS 、QPS 等，然后将指标信息进行处理，从而构建一个具有监控仪表盘、曲线图等可视化监控系统。例如，很多公司采用Kafka 与ELK (Elastic Search 、Logstash 和Kibana)整合构建应用服务监控系统。
  • 网站用户行为追踪。
  • 为了更好地了解用户行为、操作习惯，改善用户体验，进而对产品升级改进，将用户操作轨迹、内容等信息发送到Kafka 集群上，通过Hadoop 、Spark 或Strom等进行数据分析处理，生成相应的统计报告，为推荐系统推荐对象建模提供数据源，进而为每个用户进行个性化推荐。
  • 流处理。
  • 需要将己收集的流数据提供给其他流式计算框架进行处理，用Kafka 收集流数据是一个不错的选择。
  • 持久性日志。
  • Kafka 可以为外部系统提供一种持久性日志的分布式系统。日志可以在多个节点间进行备份， Kafka 为故障节点数据恢复提供了一种重新同步的机制。同时， Kafka很方便与HDFS 和Flume 进行整合，这样就方便将Kafka 采集的数据持久化到其他外部系统。


2、Kafka的安装与配置

准备三台虚拟机，分别是node01，node02，node03，并且修改hosts文件如下：

vim /etc/hosts
#注意: 前面的ip地址改成自己的ip地址
192.168.40.133 node01
192.168.40.134 node02
192.168.40.135 node03
#3台服务器的时间要一致
#时间更新：
yum install -y rdate
rdate -s  time-b.nist.gov

2.1、基础环境配置

2.1.1、JDK环境

由于Kafka 是用Scala 语言开发的，运行在JVM上，因此在安装Kafka 之前需要先安装JDK 。

安装过程略过，我这里使用的是jdk1.8。

image.png

2.1.2、ZooKeeper环境

2.1.2.1、安装ZooKeeper

Kafka 依赖ZooKeeper ，通过ZooKeeper 来对服务节点、消费者上下线管理、集群、分区元数据管理等，因此ZooKeeper 也是Kafka 得以运行的基础环境之一。

#上传zookeeper-3.4.9.tar.gz到/export/software
cd /export/software
mkdir -p /export/servers/
tar -xvf zookeeper-3.4.9.tar.gz -C /export/servers/
#创建ZooKeeper的data目录
mkdir /export/data/zookeeper -p
cd /export/servers/zookeeper-3.4.9/conf/
#修改配置文件
mv zoo_sample.cfg zoo.cfg
vim zoo.cfg
#设置data目录
dataDir=/export/data/zookeeper
#启动ZooKeeper
./zkServer.sh start
#检查是否启动成功
jps
2.1.2.3、搭建ZooKeeper集群
#在/export/data/zookeeper目录中创建myid文件
vim /export/data/zookeeper/myid
#写入对应的节点的id，如：1,2等，保存退出
#在conf下，修改zoo.cfg文件
vim zoo.cfg
#添加如下内容
server.1=node01:2888:3888
server.2=node02:2888:3888
server.3=node03:2888:3888
2.1.2.3、配置环境变量
vim /etc/profile
export ZK_HOME=/export/servers/zookeeper-3.4.9
export PATH=${ZK_HOME}/bin:$PATH
#立即生效
source /etc/profile
2.1.2.4、分发到其它机器
scp /etc/profile node02:/etc/
scp /etc/profile node03:/etc/
cd /export/servers
scp -r zookeeper-3.4.9 node02:/export/servers/
scp -r zookeeper-3.4.9 node03:/export/servers/
2.1.2.5、一键启动、停止脚本
mkdir -p /export/servers/onekey/zk
vim slave
#输入如下内容
node01
node02
node03
#保存退出
vim startzk.sh
#输入如下内容
cat /export/servers/onekey/zk/slave | while read line
do
{
 echo "开始启动 --> "$line
 ssh $line "source /etc/profile;nohup sh ${ZK_HOME}/bin/zkServer.sh start >/dev/null 2>&1 &"
}&
wait
done
echo "★★★启动完成★★★"
#保存退出
vim stopzk.sh
#输入如下内容
cat /export/servers/onekey/zk/slave | while read line
do
{
 echo "开始停止 --> "$line
 ssh $line "source /etc/profile;nohup sh ${ZK_HOME}/bin/zkServer.sh stop >/dev/null 2>&1 &"
}&
wait
done
echo "★★★停止完成★★★"
#保存退出
#设置可执行权限
chmod +x startzk.sh stopzk.sh
#添加到环境变量中
export ZK_ONEKEY=/export/servers/onekey
export PATH=${ZK_ONEKEY}/zk:$PATH
2.1.2.6、检查启动是否成功

image.png

发现三台机器都有“QuorumPeerMain”进程，说明机器已经启动成功了。

检查集群是否正常：

zkServer.sh status

image.png

image.png

image.png

发现，集群运行一切正常。

2.2、安装Kafka

2.2.1、单机版Kafka安装

第一步：上传Kafka安装包并且解压

rz 上传kafka_2.11-1.1.0.tgz到 /export/software/
cd /export/software/
tar -xvf kafka_2.11-1.1.0.tgz -C /export/servers/
cd /export/servers
mv kafka_2.11-1.1.0/ kafka

第二步：配置环境变量

vim /etc/profile
#输入如下内容
export KAFKA_HOME=/export/servers/kafka
export PATH=${KAFKA_HOME}/bin:$PATH
#保存退出
source /etc/profile

第三步：修改配置文件

cd /export/servers/kafka
cd config
vim server.properties
# The id of the broker. This must be set to a unique integer for each broker.
# 必须要只要一个brokerid，并且它必须是唯一的。
broker.id=0
# A comma separated list of directories under which to store log files
# 日志数据文件存储的路径 （如不存在，需要手动创建该目录, mkdir -p /export/data/kafka/）
log.dirs=/export/data/kafka
# ZooKeeper的配置，本地模式下指向到本地的ZooKeeper服务即可
zookeeper.connect=node01:2181
# 保存退出

第四步：启动kafka服务

# 以守护进程的方式启动kafka
kafka-server-start.sh -daemon /export/servers/kafka/config/server.properties

第五步：检测kafka是否启动

image.png

如果进程中有名为kafka的进程，就说明kafka已经启动了。

2.2.2、验证kafka是否安装成功

由于kafka是将元数据保存在ZooKeeper中的，所以，可以通过查看ZooKeeper中的信息进行验证kafka是否安装成功。

image.png

image.png

image.png

2.2.3、部署kafka-manager

Kafka Manager 由 yahoo 公司开发，该工具可以方便查看集群 主题分布情况，同时支持对 多个集群的管理、分区平衡以及创建主题等操作。

源码托管于github：https://github.com/yahoo/kafka-manager

第一步：上传Kafka-manager安装包并且解压

rz上传kafka-manager-1.3.3.17.tar.gz到 /export/software/
cd /export/software
tar -xvf kafka-manager-1.3.3.17.tar.gz -C /export/servers/
cd /export/servers/kafka-manager-1.3.3.17/conf

第二步：修改配置文件

#修改配置文件
vim application.conf
#新增项，http访问服务的端口
http.port=19000
#修改成自己的zk机器地址和端口
kafka-manager.zkhosts="node01:2181"
#保存退出

第三步：启动服务

cd /export/servers/kafka-manager-1.3.3.17/bin
#启动服务
./kafka-manager -Dconfig.file=../conf/application.conf
#制作启动脚本
vim /etc/profile
export KAFKA_MANAGE_HOME=/export/servers/kafka-manager-1.3.3.17
export PATH=${KAFKA_MANAGE_HOME}/bin:$PATH
source /etc/profile
cd /export/servers/onekey/
mkdir kafka-manager
cd kafka-manager
vim start-kafka-manager.sh
nohup kafka-manager -Dconfig.file=${KAFKA_MANAGE_HOME}/conf/application.conf >/dev/null 2>&1 &
chmod +x start-kafka-manager.sh
vim /etc/profile
export PATH=${ZK_ONEKEY}/kafka-manager:$PATH
source /etc/profile

第四步：检查是否启动成功

打开浏览器，输入地址：http://node01:19000/，即可看到kafka-manage管理界面。

image.png

2.2.4、kafka-manager的使用

进入管理界面，是没有显示Cluster信息的，需要添加后才能操作。

  • 添加 Cluster：

image.png

输入Cluster Name、ZooKeeper信息、以及Kafka的版本信息（这里最高只能选择1.0.0）。

image.png

点击Save按钮保存。

image.png

添加成功。

  • 查看kafka的信息
    image.png
  • 查看Broker信息
    image.png
  • 查看Topic列表
    image.png
  • 查看单个topic信息以及操作
    image.png
  • 优化副本选举
    image.png
  • 查看消费者信息
    image.png

2.2.5、搭建kafka集群

kafka集群的搭建是非常简单的，只需要将上面的单机版的kafka分发的其他机器，并且将ZooKeeper信息修改成集群的配置以及设置不同的broker值即可。

第一步：将kafka分发到node02、node03

cd /export/servers/
scp -r kafka node02:/export/servers/
scp -r kafka node03:/export/servers/
scp /etc/profile node02:/etc/
scp /etc/profile node03:/etc/
# 分别到node02、node03机器上执行
source /etc/profile

第二步：修改node01、node02、node03上的kafka配置文件

  • node01：
cd /export/servers/kafka/config
vim server.properties
zookeeper.connect=node01:2181,node02:2181,node03:2181
  • node02：
cd /export/servers/kafka/config
vim server.properties
broker.id=1
zookeeper.connect=node01:2181,node02:2181,node03:2181
  • node03：
cd /export/servers/kafka/config
vim server.properties
broker.id=2
zookeeper.connect=node01:2181,node02:2181,node03:2181

第三步：编写一键启动、停止脚本。注意：该脚本依赖于环境变量中的KAFKA_HOME。

mkdir -p /export/servers/onekey/kafka
vim slave
#输入如下内容
node01
node02
node03
#保存退出
vim start-kafka.sh
#输入如下内容
cat /export/servers/onekey/kafka/slave | while read line
do
{
 echo "开始启动 --> "$line
 ssh $line "source /etc/profile;nohup sh ${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties >/dev/null 2>&1 &"
}&
wait
done
echo "★★★启动完成★★★"
#保存退出
chmod +x start-kafka.sh
vim stop-kafka.sh
#输入如下内容
cat /export/servers/onekey/kafka/slave | while read line
do
{
 echo "开始停止 --> "$line
 ssh $line "source /etc/profile;nohup sh ${KAFKA_HOME}/bin/kafka-server-stop.sh >/dev/null 2>&1 &"
}&
wait
done
echo "★★★停止完成★★★"
#保存退出
chmod +x stop-kafka.sh
#加入到环境变量中
export PATH=${ZK_ONEKEY}/kafka:$PATH
source /etc/profile

第四步：通过kafka-manager管理工具查看集群信息。

image.png

由此可见，kafka集群已经启动完成。

文章标签：
云消息队列 Kafka 版
微服务引擎
日志服务
Java
C++
Python
消息中间件
搜索推荐
分布式计算
监控
Kafka
Apache
Scala
关键词：
云消息队列 Kafka 版入门
云消息队列 Kafka 版入门宝典截图
大数据流动
目录
相关文章
诸葛子房
|
5月前
|
消息中间件 Java Kafka
kafka入门demo
kafka入门demo
诸葛子房
65 0
程序员朱永胜
|
消息中间件 监控 关系型数据库
【Kafka系列】（一）Kafka入门（下）
【Kafka系列】（一）Kafka入门（下）
程序员朱永胜
122 0
孙玉洁-47170
|
5月前
|
消息中间件 分布式计算 Kafka
SparkStreaming（SparkStreaming概述、入门、Kafka数据源、DStream转换、输出、关闭）
SparkStreaming（SparkStreaming概述、入门、Kafka数据源、DStream转换、输出、关闭）（一）
孙玉洁-47170
88 5
yuanzhengme
|
5月前
|
消息中间件 Java Kafka
Kafka【环境搭建 01】kafka_2.12-2.6.0 单机版安装+参数配置及说明+添加到service服务+开机启动配置+验证+chkconfig配置说明（一篇入门kafka）
【2月更文挑战第19天】Kafka【环境搭建 01】kafka_2.12-2.6.0 单机版安装+参数配置及说明+添加到service服务+开机启动配置+验证+chkconfig配置说明（一篇入门kafka）
yuanzhengme
159 1
木易巷
|
5月前
|
消息中间件 存储 Kafka
Kafka【基础入门】
Kafka【基础入门】
木易巷
58 1
小小工匠
|
5月前
|
消息中间件 存储 分布式计算
Apache Kafka-初体验Kafka(01)-入门整体认识kafka
Apache Kafka-初体验Kafka(01)-入门整体认识kafka
小小工匠
70 0
lady_killer9
|
5月前
|
消息中间件 算法 Kafka
Kafka入门，这一篇就够了（安装，topic，生产者，消费者）
Kafka入门，这一篇就够了（安装，topic，生产者，消费者）
lady_killer9
238 0
隐-风
|
消息中间件 存储 Kafka
（四）kafka从入门到精通之安装教程
Kafka是一个高性能、低延迟、分布式的分布式数据库，可以在分布式环境中实现数据的实时同步和分发。Zookeeper是一种开源的分布式数据存储系统，它可以在分布式环境中存储和管理数据库中的数据。它的主要作用是实现数据的实时同步和分发，可以用于实现分布式数据库、分布式文件系统、分布式日志系统等。Zookeeper的设计目标是高可用性、高性能、低延迟，它支持多种客户端协议，包括TCP和HTTP，可以方便地与其他分布式系统进行集成。
隐-风
125 0
隐-风
|
消息中间件 传感器 Kafka
（三）kafka从入门到精通之使用场景
Kafka 是一种流处理平台，主要用于处理大量数据流，如实时事件、日志文件和传感器数据等。Kafka的目的是实现高吞吐量、低延迟和高可用性的数据处理。Kafka提供了一个高度可扩展的架构，可以轻松地添加和删除节点，并且能够处理数百亿条消息/分区。Kafka的消息可以容错，即使某个节点失败，消息也会在集群中的其他节点上得到处理。总的来说，Kafka 是一个非常强大的数据处理平台，可以用于实时数据处理、日志文件处理、传感器数据处理和流处理等场景。
隐-风
141 0
程序员朱永胜
|
消息中间件 存储 Java
【Kafka系列】（一）Kafka入门（上）
【Kafka系列】（一）Kafka入门
程序员朱永胜
68 0

热门文章

最新文章

  • 1
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 2
    SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS
  • 3
    springboot配置kafka生产者和消费者详解
  • 4
    Kafka实战－实时日志统计流程
  • 5
    实时数据同步与共享：使用Apache Kafka Connect
  • 6
    Kafka是什么？有什么优点和缺点？
  • 7
    flume-kafka 实例 详细
  • 8
    Apache Kafka开发入门指南
  • 9
    阿里云Kafka幂等生产者与事务生产者
  • 10
    kafka集群搭建
  • 1
    扒开kafka内部组件，咱瞅一瞅都有啥？
    60
  • 2
    DataWorks操作报错合集之通过dataworks把表数据发送到kafka，表字段全是string类型如何解决
    64
  • 3
    人工智能平台PAI 操作报错合集之pyalink 1.6.1StreamOperator.fromDataframe(out_df, out_schema_str)之后直接连kafka sink会报下面的错误如何解决
    230
  • 4
    [AIGC 大数据基础] 大数据流处理 Kafka
    44
  • 5
    Python面试：消息队列（RabbitMQ、Kafka）基础知识与应用
    149
  • 6
    Kafka 的生产者优秀架构设计
    63
  • 7
    深度探索：使用Apache Kafka构建高效Java消息队列处理系统
    99
  • 8
    【Kafka】Kafka 的日志保留期与数据清理策略
    1596
  • 9
    【Kafka】Kafka 创建Topic后如何将分区放置到不同的 Broker 中？
    255
  • 10
    【Kakfa】Kafka 的Topic中 Partition 数据是怎么存储到磁盘的？
    235

    • 相关课程

    更多
  • 消息队列Kafka入门课程
  • 分布式消息系统 Kafka 快速入门
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第一阶段）
  • Redis入门到精通（进阶篇）
  • 「大师课」深度剖析 RocketMQ5.0

    • 相关电子书

    更多
  • Java Spring Boot开发实战系列课程【第16讲】：Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 消息队列kafka介绍

    • 相关实验场景

    更多
  • 使用Flink实时发现最热Github项目
  • Arthas 进阶教程
  • Arthas 基础教程
  • Rocketmq-spring入门与实践
    • 下一篇
    无影云桌面