jjzjj

Kafka系统整理 一

6个日的梦想 2024-04-19 原文

一、Kafka 概述

1.1 定义

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

kafka最新定义:kafka是一个开源的分布式事件流平台(Event Streaming Platform), 被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

1.2 消息队列

目 前企 业中比 较常 见的 消息 队列产 品主 要有 Kafka ActiveMQ RabbitMQ
RocketMQ 等。
在大数据场景主要采用 Kafka 作为消息队列。在 JavaEE 开发中主要采用 ActiveMQ
RabbitMQ RocketMQ

1.3 应用场景

主要应用场景包括: 缓存 / 消峰 解耦 异步通信。
1. 缓冲 / 消峰: 有助于控制和优化数据流经过系统的速度,解决生产消息和消费消息的处理速度不一致的情况。
2.  解耦: 允许你独立的扩展或修改两边的处理过程,只要确保它们遵守同样的接口约束。
3.  异步通信: 允许用户把一个消息放入队列,但并不立即处理它, 然后 在需要的时候再去处理它们

1.4 消息队列的两种模式

 1.5 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

 

 二、Kafka 快速入门

先部署zookeeper

 1. 集群规划

 2. 集群部署

0 官方下载地址: http://kafka.apache.org/downloads.html
1 解压安装包
[atguigu@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C
/opt/module/

2修改解压后的文件名称

[atguigu@hadoop102 module]$ mv kafka_2.12-3.0.0/ kafka
3 进入到 /opt/module/kafka 目录,修改配置文件
[atguigu@hadoop102 kafka]$ cd config/
[atguigu@hadoop102 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
4 分发安装包

[atguigu@hadoop102 module]$ xsync kafka/
5 分别在 hadoop103 hadoop104 上修改配置文件 /opt/module/kafka/config/server.properties
中的 broker.id=1 broker.id=2
注: broker.id 不得重复,整个集群中唯一。
6 配置环境变量
1 )在 /etc/profile.d/my_env.sh 文件中增加 kafka 环境变量配置
[atguigu@hadoop102 module]$ sudo vim /etc/profile.d/my_env.sh

增加如下内容:

#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin
(2)刷新一下环境变量。
[atguigu@hadoop102 module]$ source /etc/profile
(3)分发环境变量文件到其他节点,并 source
[atguigu@hadoop102 module]$
sudo /home/atguigu/bin/xsync
/etc/profile.d/my_env.sh
[atguigu@hadoop103 module]$ source /etc/profile
[atguigu@hadoop104 module]$ source /etc/profile

7 启动集群
1 )先启动 Zookeeper 集群,然后启动 Kafka
[atguigu@hadoop102 kafka]$ zk.sh start
(2)依次在 hadoop102 hadoop103 hadoop104 节点上启动 Kafka
[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon
config/server.properties
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon
config/server.properties
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon
config/server.properties
8 关闭集群
[atguigu@hadoop102 kafka]$ bin/kafka-server-stop.sh
[atguigu@hadoop103 kafka]$ bin/kafka-server-stop.sh
[atguigu@hadoop104 kafka]$ bin/kafka-server-stop.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 )添加执行权限
[atguigu@hadoop102 bin]$ chmod +x kf.sh
3 )启动集群命令
[atguigu@hadoop102 ~]$ kf.sh start
4 )停止集群命令
[atguigu@hadoop102 ~]$ kf.sh stop
注意: 停止 Kafka 集群时,一定要等 Kafka 所有节点进程全部停止后再停止 Zookeeper
集群。因为 Zookeeper 集群当中记录着 Kafka 集群相关信息, Zookeeper 集群一旦先停止,
Kafka 集群就没有办法再获取停止进程的信息,只能手动杀死 Kafka 进程了(kill -9 xxx)。

三、主要命令记录 

1 主题命令

1 )查看操作主题命令参数
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh
2 )查看当前服务器中的所有 topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
hadoop102:9092 --list
3 )创建 first topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first
选项说明:
--topic 定义 topic
--replication-factor 定义副本数
--partitions 定义分区数
4 )查看 first 主题的详情
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
hadoop102:9092 --describe --topic first
5)修改分区数 (注意:分区数只能增加,不能减少)
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
hadoop102:9092 --alter --topic first --partitions 3
7 )删除 topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server 
hadoop102:9092 --delete --topic first

2 生产者命令

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

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh
2 )发送消息
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --
bootstrap-server hadoop102:9092 --topic first
>hello 
>world

3 消费者命令

1 )查看操作消费者命令参数
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh
2 )消费消息
(1)消费 first 主题中的数据。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --
bootstrap-server hadoop102:9092 --topic first
(2)把主题中所有的数据都读取出来(包括历史数据)。
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --
bootstrap-server hadoop102:9092 --from-beginning --topic first

有关Kafka系统整理 一的更多相关文章

  1. 电脑0x0000001A蓝屏错误怎么U盘重装系统教学 - 2

      电脑0x0000001A蓝屏错误怎么U盘重装系统教学分享。有用户电脑开机之后遇到了系统蓝屏的情况。系统蓝屏问题很多时候都是系统bug,只有通过重装系统来进行解决。那么蓝屏问题如何通过U盘重装新系统来解决呢?来看看以下的详细操作方法教学吧。  准备工作:  1、U盘一个(尽量使用8G以上的U盘)。  2、一台正常联网可使用的电脑。  3、ghost或ISO系统镜像文件(Win10系统下载_Win10专业版_windows10正式版下载-系统之家)。  4、在本页面下载U盘启动盘制作工具:系统之家U盘启动工具。  U盘启动盘制作步骤:  注意:制作期间,U盘会被格式化,因此U盘中的重要文件请注

  2. 【鸿蒙应用开发系列】- 获取系统设备信息以及版本API兼容调用方式 - 2

    在应用开发中,有时候我们需要获取系统的设备信息,用于数据上报和行为分析。那在鸿蒙系统中,我们应该怎么去获取设备的系统信息呢,比如说获取手机的系统版本号、手机的制造商、手机型号等数据。1、获取方式这里分为两种情况,一种是设备信息的获取,一种是系统信息的获取。1.1、获取设备信息获取设备信息,鸿蒙的SDK包为我们提供了DeviceInfo类,通过该类的一些静态方法,可以获取设备信息,DeviceInfo类的包路径为:ohos.system.DeviceInfo.具体的方法如下:ModifierandTypeMethodDescriptionstatic StringgetAbiList​()Obt

  3. kvm虚拟机安装centos7基于ubuntu20.04系统 - 2

    需求:要创建虚拟机,就需要给他提供一个虚拟的磁盘,我们就在/opt目录下创建一个10G大小的raw格式的虚拟磁盘CentOS-7-x86_64.raw命令格式:qemu-imgcreate-f磁盘格式磁盘名称磁盘大小qemu-imgcreate-f磁盘格式-o?1.创建磁盘qemu-imgcreate-fraw/opt/CentOS-7-x86_64.raw10G执行效果#ls/opt/CentOS-7-x86_64.raw2.安装虚拟机使用virt-install命令,基于我们提供的系统镜像和虚拟磁盘来创建一个虚拟机,另外在创建虚拟机之前,提前打开vnc客户端,在创建虚拟机的时候,通过vnc

  4. ruby - 在没有基准或时间的情况下用 Ruby 测量用户时间或系统时间 - 2

    因为我现在正在做一些时间测量,我想知道是否可以在不使用Benchmark类或命令行实用程序time的情况下测量用户时间或系统时间。使用Time类只显示挂钟时间,而不显示系统和用户时间,但是我正在寻找具有相同灵active的解决方案,例如time=TimeUtility.now#somecodeuser,system,real=TimeUtility.now-time原因是我有点不喜欢Benchmark,因为它不能只返回数字(编辑:我错了-它可以。请参阅下面的答案。)。当然,我可以解析输出,但感觉不对。*NIX系统的time实用程序也应该可以解决我的问题,但我想知道是否已经在Ruby中实

  5. ruby - 以毫秒为单位获取当前系统时间 - 2

    在Ruby中,以毫秒为单位获取自纪元(1970)以来的当前系统时间的正确方法是什么?我试过了Time.now.to_i,好像不是我想要的结果。我需要结果显示毫秒并且使用long类型,而不是float或double。 最佳答案 (Time.now.to_f*1000).to_iTime.now.to_f显示包含十进制数字的时间。要获得毫秒数,只需将时间乘以1000。 关于ruby-以毫秒为单位获取当前系统时间,我们在StackOverflow上找到一个类似的问题:

  6. ruby-on-rails - 如何构建复杂的 Rails 系统 - 2

    关闭。这个问题需要更多focused.它目前不接受答案。想改进这个问题吗?更新问题,使其只关注一个问题editingthispost.关闭8年前。Improvethisquestion我们有以下(以及更多)系统,我们将数据从一个应用推送/拉取到另一个:托管CRM(InsideSales.com)Asterisk电话系统(内部)横幅广告系统(openx,我们托管)潜在客户生成系统(自行开发)电子商务商店(spree,我们托管)工作板(本土)一些工作网站抓取+入站工作提要电子邮件传送系统(如Mailchimp,自主开发)事件管理系统(如eventbrite,自主开发)仪表板系统(大量图表和

  7. ruby-on-rails - Rails 3,在RAILS_ROOT上方显示来自本地文件系统的jpg图片 - 2

    我正在尝试找出一种方法来显示来自不在RAILS_ROOT下(在RedHat或Ubuntu环境中)的已安装文件系统的图像。我不想使用符号链接(symboliclink),因为这个应用程序实际上是通过Tomcat部署的,而当我关闭Tomcat时,Tomcat会尝试跟随符号链接(symboliclink)并删除挂载中的所有图像。由于这些文件的数量和大小,将图像放在public/images下也不是一种选择。我查看了send_file,但它只会显示一张图片。我需要在一个格式良好的页面中显示6个请求的图像。由于膨胀,我宁愿不使用Base64编码,但我不知道如何将图像数据与呈现的页面一起传递下去。

  8. ruby - 我可以从 Ruby 中的系统调用中获得连续输出吗? - 2

    当您在Ruby脚本中使用系统调用时,您可以像这样获得该命令的输出:output=`ls`putsoutput这就是thisquestion是关于。但是有没有办法显示系统调用的连续输出?例如,如果您运行此安全复制命令,以通过SSH从服务器获取文件:scpuser@someserver:remoteFile/some/local/folder/...它显示随着下载进度的连续输出。但是这个:output=`scpuser@someserver:remoteFile/some/local/folder/`putsoutput...不捕获该输出。如何从我的Ruby脚本中显示正在进行的下载进度?

  9. u盘安装系统(win10为例) - 2

    下载微PE工具箱进入官网下载微PE工具箱-下载 安装好后,打开微PE工具箱客户端,选择安装PE到U盘 PE壁纸可选择自己喜欢的壁纸,勾选上包含DOS工具箱,个性化盘符图标 下载原版系统进入网站下载镜像NEXT,ITELLYOU如果没有账号,注册一下就好进入选择开始使用选择win10 这里我们选择消费者版,用迅雷把BT种子下载下来 下面的两个盘符,是PE工具箱安装进U盘后,分成的盘符,注意EFI的盘符,这里面不能删东西,也不能添东西,另一个盘符可以当做正常的U盘空间使用,我们现在需要把下载下来的景象文件复制到正常的U盘空间中去 这个时候我们的系统U盘就只做好了 安装系统我们将U盘插入电脑,开机,

  10. Ubuntu20.04系统WineHQ7.0安装微信 - 2

    提供3种Ubuntu系统安装微信的方法,在Ubuntu20.04上验证都ok。1.WineHQ7.0安装微信:ubuntu20.04安装最新版微信--可以支持微信最新版,但是适配的不是特别好;比如WeChartOCR.exe报错。2.原生微信安装:linux系统下的微信安装(ubuntu20.04)--微信适配的最好,反应最快,但是微信版本只到2.1.1,版本太老,很多功能都没有。3.深度deepin-wine6安装微信:ubuntu20.04+系统deepin-wine6安装新版微信--综合比较好,当前个人使用此种方法1个月,微信版本3.4;没什么大问题,尚可。一、WineHQ7.0安装微信

随机推荐