【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门

这篇具有很好参考价值的文章主要介绍了【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

【Kafka-3.x-教程】专栏:

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门
【Kafka-3.x-教程】-【二】Kafka-生产者-Producer
【Kafka-3.x-教程】-【三】Kafka-Broker、Kafka-Kraft
【Kafka-3.x-教程】-【四】Kafka-消费者-Consumer
【Kafka-3.x-教程】-【五】Kafka-监控-Eagle
【Kafka-3.x-教程】-【六】Kafka 外部系统集成 【Flume、Flink、SpringBoot、Spark】
【Kafka-3.x-教程】-【七】Kafka 生产调优、Kafka 压力测试

1)定义

试想这样一个场景,某宝或某东购物平台,上游通过 Flume 采集相关日志到 Hadoop 集群中,在平时流量小的时候,Hadoop 的写入速度在 100M/s,Flume 采集日志的速度也在 100M/s,这样下游的 Hadoop 上传速度可以和上游的 Flume 采集速度相匹配,没有问题。

那么现在到了促销季,假如双十一进行大促销,那么 Flume 采集到的数据量远超过平时的 100M/s(比如达到 200M/s),而下游的 Hadoop 速度的极限已经是 100M/s 了,此时如果不解决这个问题就会导致上游数据量无限度的堆积,导致日志服务器爆掉。

此时 Kafka 出现了,试想,如果将 Flume 采集到的数据先放入一个中间栈做保存,然后让 Hadoop 慢慢消化海量数据,那么就可以解决上述的问题。

Kafka 是一个分布式的基于发布/订阅模式的消息队列(Message Queue),主要应用于大数据实时处理领域。

目前企业中比较常见的消息队列产品主要有 Kafka、ActiveMQ 、RabbitMQ 、RocketMQ 等。

在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ、RabbitMQ、RocketMQ。

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2)应用场景

2.1.削峰

缓冲/消峰:有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2.2.解耦

解耦:允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2.3.异步通信

异步通信:允许用户把一个消息放入队列,但并不立即处理它,然后在需要的时候再去处理它们

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

3)两种模式

1、点对点模式

  • 消费者主动拉取数据,消息收到后清除消息

2、发布/订阅模式

  • 可以有多个topic主题(浏览、点赞、收藏、评论等)

  • 消费者消费数据之后,不删除数据

  • 每个消费者相互独立,都可以消费到数据

3、对比

相对于点对点模式来说,发布订阅模式的优势有:

  • 消息被消费完不会被删除,可以提供给其他消费者继续消费
  • 可以有多个 topic 进行接收消息
  • 至于数据怎么删除,可以规定时间定期进行清理

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

4)基础架构

1、为方便扩展,并提高吞吐量,一个 topic 分为多个 partition。

2、配合分区的设计,提出消费者组的概念,组内每个消费者并行消费。

3、为提高可用性,为每个partition增加若干副本,类似 NameNode HA。

4、ZK 中记录谁是 leader,Kafka 2.8.0 以后也可以配置不采用 ZK。

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

(1)Producer:消息生产者,就是向 Kafka broker 发消息的客户端。

(2)Consumer:消息消费者,向 Kafka broker 取消息的客户端。

(3)Consumer Group(CG):消费者组,由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据,一个分区只能由一个组内消费者消费;消费者组之间互不影响。所有的消费者都属于某个消费者组,即消费者组是逻辑上的一个订阅者。

(4)Broker:一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic。

(5)Topic:可以理解为一个队列,生产者和消费者面向的都是一个 topic。

(6)Partition:为了实现扩展性,一个非常大的 topic 可以分布到多个 broker(即服务器)上,一个 topic 可以分为多个 partition,每个 partition 是一个有序的队列。

(7)Replica:副本。一个 topic 的每个分区都有若干个副本,一个 Leader 和若干个 Follower。

(8)Leader:每个分区多个副本的“主”,生产者发送数据的对象,以及消费者消费数据的对象都是 Leader。

(9)Follower:每个分区多个副本中的“从”,实时从 Leader 中同步数据,保持和 Leader 数据的同步。Leader 发生故障时,某个 Follower 会成为新的 Leader。

5)安装部署

5.1.集群规划

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

5.2.集群部署

1、官方下载地址

2、解压安装包

tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/

3、修改解压后的文件名称

mv kafka_2.12-3.0.0/ kafka

4、进入到/opt/module/kafka目录,修改配置文件

cd config/
vim server.properties

输入以下内容:

#broker 的全局唯一编号,不能重复,只能是数字。
broker.id=0
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka 运行日志(数据)存放的路径,路径不需要提前创建,kafka 自动帮你创建,可以
配置多个磁盘路径,路径与路径之间可以用","分隔
log.dirs=/opt/module/kafka/datas
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
# 每个 topic 创建时的副本数,默认时 1 个副本
offsets.topic.replication.factor=1
#segment 文件保留的最长时间,超时将被删除
log.retention.hours=168
#每个 segment 文件的大小,默认最大 1G
log.segment.bytes=1073741824
# 检查过期数据的时间,默认 5 分钟检查一次是否数据过期
log.retention.check.interval.ms=300000
#配置连接 Zookeeper 集群地址(在 zk 根目录下创建/kafka,方便管理)
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/ka
fka

5、分发安装包

xsync kafka/

xsync 脚本:

#!/bin/bash

#1. 判断参数个数
if [ $# -lt 1 ]
then
    echo Not Enough Arguement!
    exit;
fi

#2. 遍历集群所有机器
for host in hadoop102 hadoop103 hadoop104
do
    echo ====================  $host  ====================
    #3. 遍历所有目录,挨个发送

    for file in $@
    do
        #4. 判断文件是否存在
        if [ -e $file ]
            then
                #5. 获取父目录
                pdir=$(cd -P $(dirname $file); pwd)

                #6. 获取当前文件的名称
                fname=$(basename $file)
                ssh $host "mkdir -p $pdir"
                rsync -av $pdir/$fname $host:$pdir
            else
                echo $file does not exists!
        fi
    done
done

6、分别在 hadoop103hadoop104 上修改配置文件/opt/module/kafka/config/server.properties中的 broker.id=1、broker.id=2

注:broker.id 不得重复,整个集群中唯一。

vim kafka/config/server.properties
修改:
# The id of the broker. This must be set to a unique integer for 
each broker.
broker.id=1
[atguigu@hadoop104 module]$ vim kafka/config/server.properties
修改:
# The id of the broker. This must be set to a unique integer for 
each broker.
broker.id=2

7、配置环境变量

(1)在/etc/profile.d/my_env.sh文件中增加 kafka 环境变量配置

sudo vim /etc/profile.d/my_env.sh

增加如下内容:

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

(2)刷新一下环境变量。

source /etc/profile

(3)分发环境变量文件到其他节点,并 source。

sudo /home/atguigu/bin/xsync 
/etc/profile.d/my_env.sh
source /etc/profile
source /etc/profile

8、启动集群

(1)先启动 Zookeeper 集群,然后启动 Kafka。

zk.sh start

zk.sh 脚本:

#!/bin/bash

case $1 in
"start"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 启动 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
	done
}
;;
"stop"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 停止 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
	done
}
;;
"status"){
	for i in hadoop102 hadoop103 hadoop104
	do
		echo  ------------- zookeeper $i 状态 ------------
		ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
	done
}
;;
esac

(2)依次在 hadoop102、hadoop103、hadoop104 节点上启动 Kafka。

bin/kafka-server-start.sh -daemon config/server.properties

注意:配置文件的路径要能够到 server.properties。

9、关闭集群

bin/kafka-server-stop.sh

5.3.集群启停脚本

1、在/home/atguigu/bin目录下创建文件kf.sh脚本文件

vim kf.sh

脚本如下:

#! /bin/bash
case $1 in
"start"){
 for i in hadoop102 hadoop103 hadoop104
 do
 echo " --------启动 $i Kafka-------"
 ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -
daemon /opt/module/kafka/config/server.properties"
 done
};;
"stop"){
 for i in hadoop102 hadoop103 hadoop104
 do
 echo " --------停止 $i Kafka-------"
 ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh "
 done
};;
esac

2、添加执行权限

chmod +x kf.sh

3、启动集群命令

kf.sh start

4、停止集群命令

kf.sh stop

注意:停止 Kafka 集群时,一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper 集群。因为 Zookeeper 集群当中记录着 Kafka 集群相关信息,Zookeeper 集群一旦先停止,Kafka 集群就没有办法再获取停止进程的信息,只能手动杀死 Kafka 进程了。

6)Kafka 命令行操作

6.1.主题命令行操作

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

1、查看操作主题命令参数

bin/kafka-topics.sh

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2、查看当前服务器中的所有 topic

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list

3、创建 first topic

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first

选项说明:

–topic 定义 topic 名

–replication-factor 定义副本数

–partitions 定义分区数

4、查看 first 主题的详情

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first

5、修改分区数(注意:分区数只能增加,不能减少)

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3

6、再次查看 first 主题的详情

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first

7、删除 topic

bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

6.2.生产者命令行操作

1、查看操作生产者命令参数

bin/kafka-console-producer.sh

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2、发送消息

bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first

6.3.消费者命令行操作

1、查看操作消费者命令参数

bin/kafka-console-consumer.sh

【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门
【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门,Kafka,kafka,分布式,#kafka概述,#kafka快速入门

2、消费消息

(1)消费 first 主题中的数据。

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

(2)把主题中所有的数据都读取出来(包括历史数据)。文章来源地址https://www.toymoban.com/news/detail-819698.html

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first

到了这里,关于【Kafka-3.x-教程】-【一】Kafka 概述、Kafka 快速入门的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【大数据工具】Kafka伪分布式、分布式安装和Kafka-manager工具安装与使用

    Kafka 安装包下载地址:https://archive.apache.org/dist/kafka/ 1. Kafka 伪分布式安装 1. 上传并解压 Kafka 安装包 使用 FileZilla 或其他文件传输工具上传 Kafka 安装包: kafka_2.11-0.10.0.0.tgz 解压安装包 2. 编辑配置文件 3. 拷贝并修改配置文件 分别修改 server2.properties、server3.properties 4. 创建日志

    2024年02月14日
    浏览(48)
  • 分布式消息服务kafka

    什么是消息中间件? 消息中间件是分布式系统中重要的组件,本质就是一个具有接收消息、存储消息、分发消息的队列,应用程序通过读写队列消息来通信。 例如:在淘宝购物时,订单系统处理完订单后,把订单消息发送到消息中间件中,由消息中间件将订单消息分发到下

    2024年02月01日
    浏览(48)
  • 【分布式应用】kafka集群、Filebeat+Kafka+ELK搭建

    主要原因是由于在高并发环境下,同步请求来不及处理,请求往往会发生阻塞。比如大量的请求并发访问数据库,导致行锁表锁,最后请求线程会堆积过多,从而触发 too many connection 错误,引发雪崩效应。 我们使用消息队列,通过异步处理请求,从而缓解系统的压力。消息队

    2024年02月16日
    浏览(53)
  • 【分布式技术】消息队列Kafka

    目录 一、Kafka概述 二、消息队列Kafka的好处 三、消息队列Kafka的两种模式 四、Kafka 1、Kafka 定义 2、Kafka 简介 3、Kafka 的特性 五、Kafka的系统架构 六、实操部署Kafka集群  步骤一:在每一个zookeeper节点上完成kafka部署 ​编辑 步骤二:传给其他节点 步骤三:启动3个节点 kafka管理

    2024年01月23日
    浏览(55)
  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(50)
  • 分布式 - 消息队列Kafka:Kafka消费者的分区分配策略

    Kafka 消费者负载均衡策略? Kafka 消费者分区分配策略? 1. 环境准备 创建主题 test 有5个分区,准备 3 个消费者并进行消费,观察消费分配情况。然后再停止其中一个消费者,再次观察消费分配情况。 ① 创建主题 test,该主题有5个分区,2个副本: ② 创建3个消费者CustomConsu

    2024年02月13日
    浏览(47)
  • 分布式 - 消息队列Kafka:Kafka生产者架构和配置参数

    生产者发送消息流程参考图1: 先从创建一个ProducerRecord对象开始,其中需要包含目标主题和要发送的内容。另外,还可以指定键、分区、时间戳或标头。在发送ProducerRecord对象时,生产者需要先把键和值对象序列化成字节数组,这样才能在网络上传输。 接下来,如果没有显式

    2024年02月13日
    浏览(50)
  • 分布式 - 消息队列Kafka:Kafka生产者发送消息的方式

    不管是把Kafka作为消息队列、消息总线还是数据存储平台,总是需要一个可以往Kafka写入数据的生产者、一个可以从Kafka读取数据的消费者,或者一个兼具两种角色的应用程序。 Kafka 生产者是指使用 Apache Kafka 消息系统的应用程序,它们负责将消息发送到 Kafka 集群中的一个或多

    2024年02月13日
    浏览(44)
  • 分布式 - 消息队列Kafka:Kafka消费者和消费者组

    1. Kafka 消费者是什么? 消费者负责订阅Kafka中的主题,并且从订阅的主题上拉取消息。与其他一些消息中间件不同的是:在Kafka的消费理念中还有一层消费组的概念,每个消费者都有一个对应的消费组。当消息发布到主题后,只会被投递给订阅它的每个消费组中的一个消费者

    2024年02月13日
    浏览(45)
  • 分布式 - 消息队列Kafka:Kafka 消费者消费位移的提交方式

    最简单的提交方式是让消费者自动提交偏移量,自动提交 offset 的相关参数: enable.auto.commit:是否开启自动提交 offset 功能,默认为 true; auto.commit.interval.ms:自动提交 offset 的时间间隔,默认为5秒; 如果 enable.auto.commit 被设置为true,那么每过5秒,消费者就会自动提交 poll() 返

    2024年02月12日
    浏览(48)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包