大数据Kafka--概述

文章目录

  • 概述
    • 定义
    • 消息队列
      • 传统消息队列的应用场景
      • 消息队列的两种模式
    • Kafka基础架构
  • Kafka快速入门
    • 安装部署
      • 集群规划
      • 集群部署
      • 集群启停脚本
    • Kafka命令行操作
      • 主题命令行操作
      • 生产者命令行操作
      • 消费者命令行操作

概述

定义

在这里插入图片描述

消息队列

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

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

传统消息队列的应用场景

传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。

  • 缓存/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况
  • 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束
    在这里插入图片描述
  • 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们

消息队列的两种模式

在这里插入图片描述

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快速入门

安装部署

集群规划

hadoop102hadoop103hadoop104
zkzkzk
kafkakafkakafka

集群部署

1）官方下载地址：https://kafka.apache.org/downloads.html

2）解压安装包

[yudan@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/

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

[yudan@hadoop102 module]$ mv kafka_2.12-3.0.0/ kafka

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

[yudan@hadoop102 kafka]$ cd config/ 
[yudan@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/kafka --需要修改的参数

5）分发安装包

[yudan@hadoop102 module]$ xsync kafka/

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

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

[yudan@hadoop103 module]$ vim kafka/config/server.properties 
修改: 
# The id of the broker. This must be set to a unique integer for each broker. 
broker.id=1 [yudan@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环境变量配置

[yudan@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）刷新一下环境变量

[yudan@hadoop102 module]$ source /etc/profile

（3）分发环境变量文件到其他节点，并source。

[yudan@hadoop102 module]$ sudo /home/atguigu/bin/xsync /etc/profile.d/my_env.sh 
[yudan@hadoop103 module]$ source /etc/profile 
[yudan@hadoop104 module]$ source /etc/profile

8）启动集群

（1）先启动Zookeeper集群，然后启动Kafka。

[yudan@hadoop102   kafka]$ zk.sh start

（2）依次在hadoop102、hadoop103、hadoop104节点上启动Kafka。

[yudan@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties 
[yudan@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties 
[yudan@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties

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

9）关闭集群

[yudan@hadoop102 kafka]$ bin/kafka-server-stop.sh  
[yudan@hadoop103 kafka]$ bin/kafka-server-stop.sh  
[yudan@hadoop104 kafka]$ bin/kafka-server-stop.sh

集群启停脚本

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

[yudan@hadoop102 bin]$ 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）添加执行权限

[yudan@hadoop102 bin]$ chmod 777 kf.sh

3）启动集群命令

[yudan@hadoop102 ~]$ kf.sh start

4）停止集群命令

[yudan@hadoop102 ~]$ kf.sh stop

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

Kafka命令行操作

在这里插入图片描述

主题命令行操作

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

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh
参数描述
--bootstrap-server <String: server toconnect to>连接的 Kafka Broker 主机名称和端口号。
--topic <String: topic>操作的topic名称。
--create创建主题。
--delete删除主题。
--alter修改主题。
--list查看所有主题。
--describe查看主题详细描述。
--partitions <Integer: # of partitions>设置分区数。
--replication-factor <Integer: replication factor>设置分区副本。
--config <String: name=value>更新系统默认的配置。

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

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list

3）创建first topic

[yudan@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 主题的详情

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first

5）修改分区数（注意：分区数只能增加，不能减少）

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3

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

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first

7）删除topic

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

生产者命令行操作

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

[yudan@hadoop102 kafka]$ bin/kafka-console-producer.sh
参数描述
--bootstrap-server要连接的 Kafka Broker 主机名称和端口号。
--topic操作的topic。

2）发送消息

[yudan@hadoop102 kafka]$ bin/kafka-console-producer.sh -bootstrap-server hadoop102:9092 --topic first >hello world 
>yudan yudan

消费者命令行操作

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

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh
参数描述
--bootstrap-server<String: server toconnect to>连接的 Kafka Broker 主机名称和端口号。
--topic<String: topic> 操作的topic名称。
--from-beginning从头开始消费。如果设置此参数，消费者将从该topic的最开始位置开始消费消息
--group <String: consumer group id>指定消费者组名称

2）消费消息

（1）消费first 主题中的数据

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

（2）把主题中所有的数据都读取出来（包括历史数据）

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first

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

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

相关文章

AIoT网关 人工智能物联网网关

AIoT网关 人工智能物联网网关

AIoT(人工智能物联网)作为新一代技术的代表&#xff0c;正以前所未有的速度改变着我们的生活方式。在这个智能时代&#xff0c;AIoT网关的重要性日益凸显。它不仅是连接智能设备和应用的关键&#xff0c;同时也是实现智能化家居、智慧城市和工业自动化的必备技术。 　　 　　一…
阅读更多...
真香！NineData SQL开发全面适配 GaiaDB

真香！NineData SQL开发全面适配 GaiaDB

2 月&#xff0c;新年伊始&#xff0c;NineData 重磅发布&#xff0c;提供了对百度云原生关系型数据库 GaiaDB 的支持。 这一次的发布不仅仅是简单的数据源支持&#xff0c;而是覆盖了整个 SQL 开发能力的重要发布&#xff0c;意味着您已经可以完整地使用 NineData SQL 开发的…
阅读更多...
程序媛的mac修炼手册--小白入门Java篇

程序媛的mac修炼手册--小白入门Java篇

最近因为要用CiteSpace做文献综述&#xff0c;间接接触Java了。所以&#xff0c;继Python、C之后&#xff0c;又要涉猎Java了。刺激&#xff01;&#xff01; 由于CiteSpace与Java要求版本高度匹配&#xff0c;有个匹配详情明天为大家讲解。总之&#xff0c;我的Java之旅开始于…
阅读更多...
React18源码: reconciler执行流程

React18源码: reconciler执行流程

reconciler执行流程 1 &#xff09;概述 此处先归纳一下react-reconciler包的主要作用&#xff0c;将主要功能分为4个方面&#xff1a; 输入&#xff1a;暴露api函数&#xff08;如&#xff1a;scheduleUpdateOnFiber&#xff09;, 供给其他包&#xff08;如react包&#xff0…
阅读更多...
Eureka注册中心(黑马学习笔记)

Eureka注册中心(黑马学习笔记)

Eureka注册中心 假如我们的服务提供者user-service部署了多个实例&#xff0c;如图&#xff1a; 大家思考几个问题&#xff1a; order-service在发起远程调用的时候&#xff0c;该如何得知user-service实例的ip地址和端口&#xff1f; 有多个user-service实例地址&#xff0c…
阅读更多...
C++ 文件操作-文本文件-读取和打开文件方法详解

C++ 文件操作-文本文件-读取和打开文件方法详解

读文件步骤 #include <iostream> using namespace std; #include <fstream> #include <string> //文本文件 读文件void test(){// 1 包含头文件// 2 创建流对象ifstream ifs;// 3 打开文件 并且判断是否打开成功ifs.open("table.txt",ios::in); //…
阅读更多...
【自然语言处理】:实验5，司法阅读理解

【自然语言处理】:实验5，司法阅读理解

清华大学驭风计划课程链接 学堂在线 - 精品在线课程学习平台 (xuetangx.com) 代码和报告均为本人自己实现&#xff08;实验满分&#xff09;&#xff0c;只展示主要任务实验结果&#xff0c;如果需要详细的实验报告或者代码可以私聊博主 有任何疑问或者问题&#xff0c;也欢…
阅读更多...
通俗易懂分析：Vite和Webpack的区别

通俗易懂分析：Vite和Webpack的区别

1、对项目构建的理解 先从浏览器出发&#xff0c; 浏览器是由浏览器内核和JS引擎组成&#xff1b;浏览器内核编译解析html代码和css代码&#xff0c;js引擎编译解析JavaScript代码&#xff1b;所以从本质上&#xff0c;浏览器只能识别运行JavaScript、CSS、HTML代码。 而我们在…
阅读更多...
Vue事件处理之v-on

Vue事件处理之v-on

1. 使用及定义 定义方法 function 方法名称(接受的event或是什么都不写) {//不管方法后括号内写与不写event,都可以接受到方法内表达式 }//定义一个接受参数的方法,此时也会在传入event function 方法名称(传入参数) {//可接受传入参数与event方法内表达式 } //定义一个接受参…
阅读更多...
代码随想录算法刷题训练营day22

代码随想录算法刷题训练营day22

代码随想录算法刷题训练营day22&#xff1a;LeetCode(236)二叉树的最近公共祖先、LeetCode(235) 二叉搜索树的最近公共祖先、LeetCode(701)二叉搜索树中的插入操作、LeetCode(450)删除二叉搜索树中的节点 LeetCode(236)二叉树的最近公共祖先 题目 代码 /*** Definition for…
阅读更多...
用例图：软件设计之旅的蓝图

用例图：软件设计之旅的蓝图

用例图&#xff1a;软件设计之旅的蓝图 今天我们将探讨一个对于软件开发至关重要的主题——用例图。在软件开发过程中&#xff0c;用例图是一种静态结构图&#xff0c;用于描述系统的功能需求。它展示了参与者&#xff08;Actors&#xff09;如何与系统进行交互&#xff0c;以…
阅读更多...
MyBatis进阶

MyBatis进阶

目录 一、实现多表查询 二、#{}和${} 1、#{}和${}的使用 2、#{}和${}的区别 3、${}的使用场景 三、数据库连接池 1、数据库连接池概念 2、常见数据库连接池 3、修改连接池为Hikari 四、动态sql语句--xml 1、if标签 2、tirm标签 3、where标签 4、set标签 5、fore…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023