kafka的成神秘籍（java）

kafka的成神秘籍

kafka的简介

​ Kafka 最初是由Linkedin 即领英公司基于Scala和 Java语言开发的分布式消息发布-订阅系统，现已捐献给Apache软件基金会。Kafka 最被广为人知的是作为一个 消息队列(mq)系统存在，而事实上kafka已然成为一个流行的分布式流处理平台。其具有高吞吐、低延迟的特性，许多大数据处理系统比如storm、spark、flink等都能很好地与之集成。按照Wikipedia上的说法，kafka的核心数据结构本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”。总的来讲，kafka通常具有3重角色：

消息系统

​ Kafka和传统的消息队列比如RabbitMQ、RocketMQ、ActiveMQ类似，支持流量削锋、服务解耦、异步通信等核心功能。

流处理平台

​ Kafka 不仅能够与大多数流式计算框架完美整合，并且自身也提供了一个完整的流式处理库，即kafka Streaming。kafka Streaming提供了类似Flink中的窗口、聚合、变换、连接等功能。

存储系统

​ 通常消息队列会把消息持久化到磁盘，防止消息丢失，保证消息可靠性。Kafka的消息持久化机制和多副本机制使其能够作为通用数据存储系统来使用

​ 一句话概括：Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），在业界主要应用于大数据实时处理领域。

kafka体系结构

Topic(主题)

  • 适用于存储消息的逻辑概念，1个Topic，可以看做是消息的集合。
  • 这个消息的集合（Topic），可以接收多个生产者（Producer）推送（Push）过来的消息，也可以让多个消费者（Consumer）从中消费（Pull）消息。

分区（Partition)

  • 分区的概念，可以理解为Topic的子集：1个Topic可能只有1个分区，也可能有多个分区。这里说的多个分区，一个分区就代表磁盘的一块连续的位置，不同的分区也就是磁盘上不同的区域块。

  • 分区存在的意义：

​ 通过不同的区域块，kafka存储在Topic里的消息就可以在多个地方存储，也就是对kafka进行了水平扩展，这样可以增加kafka的 并行处理能力。

  • 对于不同分区的理解

​ 同一个Topic下，多个区域块，这些不同的区域块，存储的消息是不同的，也就是说，A、B两个区域块，不会同时存储1类消 息。

  • 对于同一个分区中的消息：

    ​ 在区块A中接收到消息时，该消息会接收它在这块区域块A中的Offset(唯一编号)，这个Offset使得kafka确定了消息在区域块中的顺序。（这样就保证了磁盘的顺序读写，减少磁盘IO）

    image-20241002131656296

Log

  • 分区在逻辑上，对应一个Log，当生产者将消息写入分区的时候，实际上就是写入一个log。

  • Log：一个逻辑概念，对应着磁盘上的一个文件夹。

  • Log的组成：由多个Segment组成，每1个Segment对应1个日志文件和一个索引文件。

Broker

​ 区域块A会给他接收的消息分配一个offset，并且x会保存到A所在的磁盘区域上。而这个功能，就是由Broker完成的。

  • Broker：1个Broker就是一个单独的kafka server。

  • Broker的主要工作：接收生产者发送来的消息,分配offset,然后将包装过的数据保存到磁盘上。

  • Broker的其他作用：接收消费者Consumer和其他Broker的请求,根据请求的类型进行相应的处理然后返回响应。

  • 这里引出集群（Cluster）的概念，1个Cluster是由多个Broker构成的，也就是说，1个Broker不会对外提供服务，而是通过Cluster的形式对外提供服务：

因为，一个Cluster里，需要1个Broker担任Controller，这个Broker就是这个集群的指挥中心，负责：

  • 管理各个分区（Partition)的状态；
  • 管理每个分区（Partition）的副本的状态；
  • 监听zookeeper的数据变化。

其他的Broker均是通过这个Controller进行指挥的，完成各自相应的功能。

关于Cluster的一主多从实现：

除了担任Controller的Broker会监听其他Broker的状态，其他Broker也会监听Controller的状态，当Controller出翔了故障，就会重新选取新的Broker担任Controller

消息

kafka中最基本的消息单元。有一串字节组成，主要由key和value构成（即key、value都是字节数组）

  • key：主要作用是 根据一定策略，将这个消息路由到制定分区中 ==> 这样就保障了，包含同一个key的消息全部写入一个分区A，不会写入另外一个分区。（即实现了Partition分区中提到的“不同分区存储的消息是不一样的”）

副本

kafka会对消息进行冗余备份，每一个分区Partition，都可以有多个副本（每一个副本包含的消息是相同的，但是不能保证同一时刻下完全相同）。

副本类型：Leader、Follower。

副本选举策略（即选举1个Leader，其余为Follower）：

  • 当分区只有1个副本，这个副本就是Leader，没有Follower。
  • 在其他不同场景，会采取不同的选举策略。
  • Leader：处理kafka中所有的读写请求
  • Follower：仅仅把数据从Leader中拉取到本地，同步更新到自己的Log中。

生产者

产生消息的对象，产生消息之后，将消息按照一定的规则推送到Topic的分区中

消费者

从Topic中拉取消息，并对消息进行消费。

  • Consumer：有一个作用，维护它消费到分区（Partition）上的什么位置（即offset的值）。
  • Consumer Group: 在kafka中，多个Consumer可以组成1个Consumer Group，1个Consumer只属于1个Consumer Group.

Consumer Group的作用：保证了这个Consumer Group订阅的Topic（Partition的集合）中的每一个分区（Partition），只被Consumer Group中的一个Consumer处理。

当然，如果要实现消息的广播消费，则将同1条消息放在多个不同的Consumer Group中即可。

就上述这个Consumer和Partition的关系可以理解下面的说法：

通过向Consumer Group中动态添加适量的Consumer， 可以触发kafka的Rebalance操作（重新分配Partition和Consumer的一一对应关系，结合Topic部分的理解，这样就实现了kafka的水平扩展能力）。

kafka的安装与使用

安装

本文采用docker安装需要确定linux主机上是否有docker环境

docker -v

image-20241002221756035

拉取zookeeper镜像与kafka镜像

#拉取zookeeper镜像
docker pull zookeeper
#拉取kafka镜像
docker pull wurstmeister/kafka

运行zookeeper

docker run --restart=always -d -p 2181:2181 --name zookeeper zookeeper
#-restart=always 停机不断重启
#-p 2181:2181 docker容器端口映射到主机的端口

运行kafka

docker run --restart=always -d --name kafka -p 19092:19092 -e KAFKA_ZOOKEEPER_CONNECT=localhost:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:19092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:19092 -e KAFKA_PORT=19092 wurstmeister/kafka
#KAFKA_ZOOKEEPER_CONNECT=localhost:2181 需要注册的 zookeeper的ip和端口

创建topic

  • 进入容器目录
    docker exec -it kafka bash
  • 执行创建topic命令

sh /opt/kafka_2.13-2.8.1/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:19092 --replication-factor 1 --partitions 1 --topic data-transform

image-20241002222929452

使用（java）

添加项目依赖
  <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
编写配置文件
spring:kafka:bootstrap-servers: localhost:9092# 消费者consumer:#分组idgroup-id: my-group# 从一次消费过信息的下一条开始消费auto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer#        配置kafka的信息的生产者producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializertemplate:# 默认的消费主题default-topic: data-transform
生产者

@Component
@Slf4j
public class DataProductService {@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public final static String TOPIC = "data-transform";public void sendDataMessage(String message){log.info("生产者生产数据开始:", DateUtils.format(new Date()),"yyyyMMdd HH:mm:ss");kafkaTemplate.send(TOPIC, message).completable().whenComplete((result, ex) -> {if (ex == null) {RecordMetadata metadata = result.getRecordMetadata();log.info("生产者生产数据成功:,数据为:{}",metadata);} else {log.error("生产者生产数据失败:", ex);}});}
}
消费者

/*** @Description:* @author:<a href="2358853434@qq.com"></a> zh* @Create : 2024/10/2**/
@Component
@Slf4j
public class DataConsumerService {@KafkaListener(topics = DataProductService.TOPIC,groupId = "my-group")public void ConsumerData(String message){log.info("消费者消费数据,数据为:{}",message);}
}
测试用例
@SpringBootTest(classes = application.class)
public class demo {@ResourceDataProductService dataProductService;@ResourceDataConsumerService dataConsumerService;@Testpublic void sendDataMessage(){long l = System.currentTimeMillis();//模拟电商下单信息推送for (int i = 1000; i < 10001; i++) {String message = "{\"orderId\":\""+i+"\",\"userId\":\""+i+"\",\"productId\":\""+i+"\",\"productName\":\""+i+"\",\"productPrice\":\""+i+"\"}";dataProductService.sendDataMessage(message);dataConsumerService.ConsumerData(message);}System.out.println(System.currentTimeMillis()-l);}
}
结果

image-20241002235126191

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

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

相关文章

【mmengine】配置器(config)（进阶）继承与导出，命令行修改配置

【mmengine】配置器(config)（进阶）继承与导出，命令行修改配置

一、配置文件的继承 1.1 继承机制概述 新建optimizer_cfg.py: optimizer dict(typeSGD, lr0.02, momentum0.9, weight_decay0.0001)新建runtime_cfg.py: device "cuda" gpu_ids [0, 1] batch_size 64 epochs 100 num_workers 8新建resnet50.py: _base_ […
阅读更多...
PostgreSQL 小课专栏大纲

PostgreSQL 小课专栏大纲

PostgreSQL 介绍及发展历史 介绍 发展历史 克隆 PostgreSQL 的源代码 找到其第一次提交记录 看下第一次提交的信息 发布策略、版本号及生命周期 探索 PostgreSQL 中的术语 总结 PostgreSQL 安装 编译方式安装 yum 方式安装 apt 方式安装 Docker 方式运行 Mac/Win…
阅读更多...
Observer(观察者模式)

Observer(观察者模式)

1. 意图 定义对象间的一种一对多的依赖关系&#xff0c;当一个对象的状态发生改变时&#xff0c;所有依赖于它的对象都得到通知并被自动更新。 在观察者模式中&#xff0c;有两类对象&#xff1a;被观察者&#xff08;Subject&#xff09;和观察者&#xff08;Observer&#xf…
阅读更多...
单片机原理与技术

单片机原理与技术

文章目录 一、单片机概述二、单片机的基本原理1. **硬件结构**2. **指令系统** 三、单片机的关键技术1. **编程技术**2. **接口技术**3. **嵌入式系统技术** 四、单片机在现代科技中的应用 一、单片机概述 单片机&#xff0c;全称为单片微型计算机&#xff08;Single-Chip Mic…
阅读更多...
Vue 路由设置

Vue 路由设置

为了防止遗忘&#xff0c;记录一下用Vue写前端配置路由时的过程&#xff0c;方便后续再需要用到时回忆。 一、举个例子 假如需要实现这样的界面逻辑&#xff1a; 在HomePage中有一组选项卡按钮用于导航到子页面&#xff0c;而子页面Page1中有一个按钮&#xff0c;其响应事件是…
阅读更多...
笔记-stm32移植ucos

笔记-stm32移植ucos

文章目录 一、UCOS的基础知识1.1 前后台系统:1.2 RTOS系统可剥夺型内核:前后台系统和RTOS系统 1.3 UCOS系统简介学习方法 二、ucossii移植Step1&#xff1a;在工程中建立存放UCOSS代码的文件夹UCOSIIStep2:向CORE文件夹添加文件Step3:向Config文件夹添加文件Step4:向port文件夹…
阅读更多...
LLM4Rec最新工作: 字节发布用于序列推荐的分层大模型HLLM

LLM4Rec最新工作: 字节发布用于序列推荐的分层大模型HLLM

前几个月 Meta HSTU 点燃各大厂商对 LLM4Rec 的热情&#xff0c;一时间&#xff0c;探索推荐领域的 Scaling Law、实现推荐的 ChatGPT 时刻、取代传统推荐模型等一系列话题让人兴奋&#xff0c;然而理想有多丰满&#xff0c;现实就有多骨感&#xff0c;尚未有业界公开真正复刻 …
阅读更多...
后端复习资料

后端复习资料

后端面试资料 语雀-图灵面试题 语雀-javaGuide 小林coding&#xff1a;https://www.xiaolincoding.com/ 面试鸭&#xff1a; https://www.mianshiya.com/ codehot香菜哥&#xff1a;https://codehot.cn/markdown/welfare 面试题&#xff1a; vizard自行整理 牛客Golang社招面…
阅读更多...
vscode中配置python虚拟环境

vscode中配置python虚拟环境

python虚拟环境作用 Python虚拟环境允许你为每个独立的项目创建一个隔离的环境&#xff0c;这样每个项目都可以拥有自己的一套Python安装包和依赖&#xff0c;不会互相影响。实际使用中&#xff0c;可以在vscode或pycharm中使用虚拟环境。 1.创建虚拟环境的方法&#xff1a; …
阅读更多...
【NLP自然语言处理】01-基础学习路径简介

【NLP自然语言处理】01-基础学习路径简介

目的&#xff1a;让大家能够在 AI-NLP 领域由基础到入门具体安排&#xff1a; NLP介绍 文本预处理RNN 及其变体&#xff08;涉及案例&#xff09;Transformer 原理详解迁移学习 和 Bert 模型详解 &#xff08;涉及案例&#xff09;特点&#xff1a; 原理 实践每个文章会有练习…
阅读更多...
04-SpringBootWeb案例（中）

04-SpringBootWeb案例（中）

3. 员工管理 完成了部门管理的功能开发之后&#xff0c;我们进入到下一环节员工管理功能的开发。 基于以上原型&#xff0c;我们可以把员工管理功能分为&#xff1a; 分页查询&#xff08;今天完成&#xff09;带条件的分页查询&#xff08;今天完成&#xff09;删除员工&am…
阅读更多...
算法题总结（十）——二叉树上

算法题总结（十）——二叉树上

#二叉树的递归遍历 // 前序遍历递归LC144_二叉树的前序遍历 class Solution {public List<Integer> preorderTraversal(TreeNode root) {List<Integer> result new ArrayList<Integer>(); //也可以把result 作为全局变量&#xff0c;只需要一个函数即可。…
阅读更多...
Linus Torvalds 要求内核开发人员编写更好的 Git 合并提交信息

Linus Torvalds 要求内核开发人员编写更好的 Git 合并提交信息

昨天在宣布 Linux 6.12-rc2 内核时&#xff0c;Linus Torvalds 要求内核维护者在提交信息方面做得更好。Torvalds 尤其希望内核维护者在描述拉取请求中的变更时&#xff0c;能更好地使用积极、命令式的语气。 Linux创建者在6.12-rc2 公告中解释道&#xff1a; 总之&#xff0c…
阅读更多...
论文阅读笔记-XLNet: Generalized Autoregressive Pretraining for Language Understanding

论文阅读笔记-XLNet: Generalized Autoregressive Pretraining for Language Understanding

前言 Google发布的XLNet在问答、文本分类、自然语言理解等任务上都大幅超越BERT，XLNet提出一个框架来连接语言建模方法和预训练方法。我们所熟悉的BERT是denoising autoencoding模型，最大的亮点就是能够获取上下文相关的双向特征表示，所以相对于标准语言模型（自回归）的预…
阅读更多...
数据结构-基于不同策略的英文单词的词频统计和检索系统

数据结构-基于不同策略的英文单词的词频统计和检索系统

数据结构课程设计：基于不同策略的英文单词的词频统计和检索系统 - 海月CSDN - 博客园 数据结构-基于不同策略的英文单词的词频统计和检索系统-CSDN博客 李冬梅老师课程设计 多学多练习！！！！ C/C++在线餐馆预订管理系统_数据结构美团餐馆预订管理信息系统-CSDN博客 高校…
阅读更多...
Java之Native详解

Java之Native详解

在Java编程中&#xff0c;Native方法是一种特殊的方法&#xff0c;它允许Java代码直接调用用其他编程语言&#xff08;如C或C&#xff09;编写的代码。这种机制使得Java程序能够与底层系统资源或其他编程语言进行交互&#xff0c;实现更高效的操作或访问特定功能。本文将详细介…
阅读更多...
计算机科学各种协议

计算机科学各种协议

tls&#xff1a;TLS” 即 “Transport Layer Security”&#xff08;传输层安全&#xff09; ssl&#xff1a;SSL&#xff08;Secure Sockets Layer&#xff0c;安全套接层&#xff09;协议 Qos&#xff1a;网络服务质量&#xff08;Quality of Service&#xff0c;QoS&#…
阅读更多...
前端的全栈混合之路Meteor篇：关于前后端分离及与各框架的对比

前端的全栈混合之路Meteor篇：关于前后端分离及与各框架的对比

这篇属于番外&#xff0c;属于技术性的讨论文&#xff0c;主要谈一下可能困惑不少人的问题。meteor看似一个前后端混合的框架&#xff0c;但实际上它并不是前后端混合的&#xff0c;只是共享了一个数据结构&#xff08;数据对象&#xff09;。现实中很多团队都说是前后端分离的…
阅读更多...
【AIGC】ChatGPT提示词Prompt高效编写模式：结构化Prompt、提示词生成器与单样本/少样本提示

【AIGC】ChatGPT提示词Prompt高效编写模式：结构化Prompt、提示词生成器与单样本/少样本提示

博客主页&#xff1a; [小ᶻZ࿆] 本文专栏: AIGC | ChatGPT 文章目录 &#x1f4af;前言&#x1f4af;结构化Prompt (Structured Prompt)组成元素应用实例优势结论 &#x1f4af;提示词生成器 (Prompt Creator)如何工作应用实例优势结论 &#x1f4af;单样本/少样本提示 (O…
阅读更多...
什么是安全运营中心 SOC？

什么是安全运营中心 SOC？

SOC 代表安全运营中心&#xff0c;它是任何企业中负责组织安全、保护企业免受网络风险的单一、集中的团队或职能。 安全运营中心将管理和控制业务运营的所有安全要素&#xff0c;从监控资产到雇用合适的人员和流程&#xff0c;再到检测和应对威胁。 在本文中&#xff0c;我们…
阅读更多...
最新文章

Copyright @ 2022~2023