简单回顾一下kafka的学习

news2024/7/4 4:46:25

简单回顾一下kafka的学习

    • What
      • Broker
      • Controller
      • Partition
      • Replication
      • Topic
      • Producer
      • Consumer
    • Why
      • 为什么有多个分区
      • 为什么有副本
    • How
      • 搭建集群
      • Java简单使用
        • Producer
        • Consumer
          • offset提交方式
            • 自动提交 - 默认
            • 手动提交
          • 消费者poll消息的过程
          • 指定分区消费
          • 消息回溯消费
          • 指定offset消费
          • 新消费组的消费偏移量
    • 重点
      • Controller、Rebalance和HW
        • Controller
        • Rebalance
        • HW和LEO
      • Kafka线上问题优化
        • 如何防止消息丢失
        • 如何防止消息的重复消费
        • 幂等性如何保证
        • 如何做到顺序消费RocketMQ
        • 解决消息积压问题
        • 延迟队列
    • 附录
      • Kafka-eagle
      • 学习地址

What

Broker

在Kafka中，Broker是指Kafka集群中的一个节点（服务器）或一个实例。每个Broker都是独立运行的Kafka服务器，负责接收、存储和转发消息。它们协同工作以构建一个分布式、高可靠性的消息系统。

Controller

Controller是Kafka集群中的一种特殊角色，每个时刻只有一个Broker扮演Controller角色。Controller负责管理整个Kafka集群的元数据信息，包括维护Topic
、Partition的状态以及监控Broker的健康状态。当发生故障或者新的Broker加入集群时，Controller会负责重新分配Partition的副本。

Partition

Partition是Kafka中对消息进行分区的概念。一个Topic可以被划分为多个Partition，每个Partition是一个有序且持久化存储的消息日志。每条消息都被写入到一个特定的Partition，并且每个Partition可以在不同的Broker上进行复制，以提供高可用性和容错性。

Replication

在Kafka中，消息的副本复制是为了提供高可用性和数据冗余。每个Partition可以配置多个副本，其中一个副本称为Leader，其余的副本称为Follower。Leader负责处理读写请求，而Follower则通过复制Leader的消息来提供备份和容错能力。如果Leader发生故障，Controller将从Follower中选举出新的Leader。

Topic

Topic是Kafka中消息发布和订阅的逻辑名称。它是一个消息的分类或者主题。生产者（Producer）将消息发送到特定的Topic，而消费者（Consumer）则订阅感兴趣的Topic来接收消息。每个Topic可以划分为多个Partition，并且可以在多个Broker上进行复制以实现高可用性。

Producer

消息的生产者

Consumer

消息的消费者，同一个消费组如果是在生产者单播的情况下，就只能由一个消费者进行消费，防止重复消费

这些概念之间的关系如下：一个Kafka集群由多个Broker组成，每个Broker负责存储和传输消息。集群中有一个Controller负责管理元数据和协调操作。每个Topic可以划分为多个Partition，每个Partition可以在多个Broker上进行复制。这种分区和复制的机制提供了高吞吐量、可扩展性和容错性的特性，使得Kafka成为一种强大的分布式消息系统。

在这里插入图片描述

Why

为什么有多个分区

  1. 同一个主题有了多个分区，那么我就可以多个生产者并行的写到同一个的主题的不同分区。显然能提高写的吞吐量；
    同理消费的时候也可从同一个主题的不同分区并行消费，提高读的吞吐量。
  2. 如果同一主题消息过大过多，那么分区可以解决存储文件过大的问题。

为什么有副本

防止leader挂了之后，无法提供服务；如果副本（follower）存在就可以成为新的leader提供数据服务。

How

搭建集群

# zookeeper进行节点管理、监控
docker pull wurstmeister/zookeeper

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

# kafka集群搭建
docker search kafka --limit=50
docker pull wurstmeister/kafka

docker run -d --name kafka0 -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=master:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://master:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

docker run -d --name kafka1 -p 9093:9093 -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=master:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://master:9093 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9093 -t wurstmeister/kafka

docker run -d --name kafka2 -p 9094:9094 -e KAFKA_BROKER_ID=2 -e KAFKA_ZOOKEEPER_CONNECT=master:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://master:9094 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9094 -t wurstmeister/kafka

Java简单使用

Producer

AsyncProducer：异步发送

SyncProducer：同步发送

在同步发消息的场景下：生产者发动broker上后，ack会有 3 种不同的选择：

  1. acks=0：
    表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
  2. acks=1：
    至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
  3. acks=-1或all： 需要等待 min.insync.replicas(默认为 1 ，推荐配置大于等于2)
    这个参数配置的副本个数都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

重试机制 & 消息缓冲区

  • 发送会默认会重试 3 次，每次间隔100ms
  • 发送的消息会先进入到本地缓冲区（32mb），kafka会跑一个线程，该线程去缓冲区中取16k的数据，发送到kafka，如果到
    10 毫秒数据没取满16k，也会发送一次。
   //批次大小
   properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"16384"); 
   //多长时间发送一个批次
   properties.put(ProducerConfig.LINGER_MS_CONFIG,"1");
   //最大缓存
   properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432")

Consumer

CustomerConsumer：简单的消费者

TimeConsumer：按照时间进行消费

offset提交方式
自动提交 - 默认
```java
   // 是否自动提交offset，默认就是true
   props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
   // 自动提交offset的间隔时间
   props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
```
消费者poll到消息后默认情况下，会自动向broker的_consumer_offsets主题提交当前主题-分区消费的偏移量。

自动提交会丢消息： 因为如果消费者还没消费完poll下来的消息就自动提交了偏移量，那么此 时消费者挂了，于是下一个消费者会从已提交的offset的下一个位置开始消费消息。之前未被消费的消息就丢失掉了
手动提交
  • 手动同步提交
    if (records.count() > 0 ) {
       // 手动同步提交offset，当前线程会阻塞直到offset提交成功
       // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
       consumer.commitSync();
    }
  • 手动异步提交
     if (records.count() > 0 ) {
         // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
         consumer.commitAsync(new OffsetCommitCallback() {
             @Override
                 public void onComplete(Map<TopicPartition, OffsetAndMetadata>offsets, Exception exception) {
                     if (exception != null) {
                         System.err.println("Commit failed for " + offsets);
                         System.err.println("Commit failed exception: " +exception.getStackTrace());
                 }
             }
         });
     }
消费者poll消息的过程
  • 消费者建立了与broker之间的⻓连接，开始poll消息。
  • 默认一次poll 500条消息
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500 );

可以根据消费速度的快慢来设置，因为如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能力过弱，将其踢出消费组。将分区分配给其他消费者。可以通过这个值进行设置：

props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000 );

每次poll的时间为1s。

  • 如果1s内直接poll到500条消息那么直接消费消息并进入下一轮poll；
  • 如果1s内没有poll到500条消息，则继续去poll消息，循环往复直到累计500条或者到了1s。
    • 如果poll到消息了，就开始消费1s内poll到的消息。
    • 如果没有poll到就直接进入下一轮poll
ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));

消费者发送心跳的时间间隔

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000 );

kafka如果超过 10
秒没有收到消费者的心跳，则会把消费者踢出消费组，进行rebalance，把分区分配给其他消费者。

props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000 );
指定分区消费
consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
消息回溯消费
consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0 )));
指定offset消费
consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seek(new TopicPartition(TOPIC_NAME, 0 ), 10 );
新消费组的消费偏移量

当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费?

  • latest(默认) ：只消费自己启动之后发送到主题的消息
  • earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

重点

Controller、Rebalance和HW

Controller

Kafka集群中的broker在zk中创建临时序号节点，序号最小的节点（最先创建的节点）将作为集群的controller，负责管理整个集群中的所有分区和副本的状态：

  • 当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
  • 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
  • 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

Rebalance

该机制前提是：消费者没有指明分区消费。当消费组里消费者和分区的关系发生变化，那么就会触发rebalance机制。

这个机制会重新调整消费者消费哪个分区。在触发rebalance机制之前，消费者消费哪个分区有三种策略：

  • range：通过公式来计算某个消费者消费哪个分区
  • 轮询：大家轮着消费
  • sticky：在触发了rebalance后，在消费者消费的原分区不变的基础上对剩下的分区进行调整。

HW和LEO

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer
最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息，
consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证
了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。

在这里插入图片描述

Kafka线上问题优化

如何防止消息丢失

  • 发送方： ack是 1 或者-1/all
    可以防止消息丢失，如果要做到99.9999%，ack设成all，把min.insync.replicas配置成副本大于等于2
  • 消费方：把自动提交改为手动提交。
  • 定时任务 && 人工

如何防止消息的重复消费

一条消息被消费者消费多次。如果为了消息的不重复消费，而把生产端的重试机制关闭、消费端的手动提交改成自动提交，这样反而会出现
消息丢失，那么可以直接在防治消息丢失的手段上再加上消费消息时的幂等性保证，就能解决消息的重复消费问题。

幂等性如何保证

  • mysql 插入业务id作为主键，主键是唯一的，所以一次只能插入一条
  • 使用redis或zk的分布式锁（主流的方案）

如何做到顺序消费RocketMQ

  • 发送方：在发送时将ack不能设置 0
    ，关闭重试，使用同步发送，等到发送成功再发送下一条。确保消息是顺序发送的。
  • 接收方：消息是发送到一个分区中，只能有一个消费组的消费者来接收消息。因此，kafka的顺序消费会牺牲掉性能。

解决消息积压问题

消息积压会导致很多问题，比如磁盘被打满、生产端发消息导致kafka性能过慢，就容易出现服务雪崩，就需要有相应的手段：

  1. 在一个消费者中启动多个线程，让多个线程同时消费来提升一个消费者的消费能力（增加分区增加消费者）
  2. 如果方案一还不够的话，这个时候可以启动多个消费者，多个消费者部署在不同的服务器上。其实多个消费者部署在同一服务器上也可以提高消费能力来充分利用服务器的cpu资源。
  3. 让一个消费者去把收到的消息往另外一个topic上发，另一个topic设置多个分区和多个消费者
    ，进行具体的业务消费

延迟队列

延迟队列的应用场景：在订单创建成功后如果超过 30
分钟没有付款，则需要取消订单，此时可用延时队列来实现.

创建多个topic，每个topic表示延时的间隔

topic_5s: 延时5s执行的队列

topic_1m: 延时 1 分钟执行的队列

topic_30m: 延时 30 分钟执行的队列

发送者发送消息到相应的topic，并带上消息的发送时间;消费者订阅相应的topic，消费时轮询消费整个topic中的消息

  • 如果消息的发送时间，和消费的当前时间超过预设的值，比如 30 分钟 则进行消费
  • 如果消息的发送时间，和消费的当前时间没有超过预设的值，则不消费当前的offset及之后的offset的所有消息都消费。下次继续消费该offset处的消息，判断时间是否已满足预设值

附录

Kafka-eagle

  1. 安装Kafka-eagle 官网下载压缩包 http://www.kafka-eagle.org/

  2. 安装jdk

  3. 解压缩后修改配置文件system-config.properties

# 配置zk  去掉cluster2efak.zk.cluster.alias=cluster1cluster1.zk.list=172.16.253.35:2181
# cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181

# 配置mysql
kafka.eagle.driver=com.mysql.cj.jdbc.Driver
kafka.eagle.url=jdbc:mysql://master:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password= 123456
  1. 修改/etc/profile
export  JAVA_HOME=/usr/local/jdk/jdk1.8.0_191
CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
export KE_HOME=/home/aisys/efak-web-2.0.9
export PATH=$PATH:$KE_HOME/bin:$JAVA_HOME/bin
  1. 刷新配置
source /etc/profile
  1. 进入到bin目录，为ke.sh增加可执行的权限
chmod +x ke.sh
  1. 启动kafka-eagle
./ke.sh start

学习地址

kafka-learning

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

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

相关文章

Firefly

Firefly

Firefly(流萤): 中文对话式大语言模型在本文中&#xff0c;笔者将介绍关于Firefly&#xff08;流萤&#xff09;模型的工作&#xff0c;一个中文对话式大语言模型。https://mp.weixin.qq.com/s/TX7wj8IzD_EaMTvk0bjRtA一个支持中文的176B开源基础模型BLOOM&#xff1a;从数据源…
阅读更多...
git merge 和git rebase的区别

git merge 和git rebase的区别

文章目录 1. 概念2. git merge2.1. 示例 3. git rebase3.1. 示例 4. 总结 1. 概念 在Git版本控制系统中&#xff0c;有两种方式可以将一个分支的更改合并到另一个分支&#xff1a;git merge 和 git rebase。虽然它们都可以完成相同的任务&#xff0c;但它们的实现方式有所不同…
阅读更多...
faster-rcnn.pytorch项目环境配置（从0到1）

faster-rcnn.pytorch项目环境配置（从0到1）

faster-rcnn.pytorch项目环境配置&#xff08;从0到1&#xff09; 其实pytorch版本和CUDA版本高&#xff0c;都没有关系&#xff01;&#xff01;&#xff01;都可以适配&#xff0c;显卡30系、20系都没关系&#xff0c;都可以用&#xff01; 下面我将在AutoDL平台上&#xf…
阅读更多...
Mapbox 实现热力图教程

Mapbox 实现热力图教程

热力图在 maobox 中属于专题图的一种，他通过点的颜色和权重 来渲染点和点周围的指标情况。本文来跟大家分享一下如何使用 maobox 实现热力图的功能。 我们以全国十大名茶产区的温度指标为例，来做一个像上图这样的效果， 首先要有相关的点数据： var tentea = {type: &qu…
阅读更多...
某制药CDMO头部企业IPD变革项目启动会顺利召开

某制药CDMO头部企业IPD变革项目启动会顺利召开

近日&#xff0c;某制药CDMO头部企业IPD流程变革项目在重庆顺利召开。为进一步优化公司研发管理体系、梳理正向研发理念及文化、培养专业人才&#xff0c;经过深入调研后&#xff0c;某制药CDMO头部企业聘请科济管线首席顾问江新安教授及其团队&#xff0c;为该企业作IPD项目变…
阅读更多...
EMAS热修复Sophix适配App加固的技术方案

EMAS热修复Sophix适配App加固的技术方案

一、问题描述 某阿里云EMAS客户的APK基于最新的线上版本发布了第十个补丁&#xff0c;发布1小时后在崩溃检测平台收集到crash日志&#xff0c;并收到用户反馈&#xff1a;部分手机上的APP会闪退。 客户开发人员紧急上报EMAS技术支持&#xff0c;双方沟通后初步判断是补丁加载…
阅读更多...
使用rewriteBatchedStatements属性优化Mybatis-Plus批量插入数据

使用rewriteBatchedStatements属性优化Mybatis-Plus批量插入数据

前言 由于项目是使用MyBatis-Plus开发的&#xff0c;用起来也确实比较方便&#xff0c;尤其是service层封装好的一些通用的增删改查方法&#xff0c;省去了不少sql语句的书写&#xff0c;但是在开发过程中&#xff0c;我也发现MyBatis-Plus的saveBatch批量插入方法针对MySQL数…
阅读更多...
MAYA横纵拉窗帘

MAYA横纵拉窗帘

设置移动动画 向上拉帘
阅读更多...
File System Access API 浅析

File System Access API 浅析

前言 最近在用python&#xff08;tkinter GUI库&#xff09;做一个小工具时&#xff0c;选择文件后可以获得其真实路径。而前端&#xff08;浏览器&#xff09;出于安全和性能等方面的考虑&#xff0c;对文件的操作是非常局限的。 在HTML5标准的File API之前&#xff0c;纯前…
阅读更多...
【Redis】高可用之复制（replica）

【Redis】高可用之复制（replica）

本文是Redis系列第4篇&#xff0c;前3篇欢迎移步 【Redis】不卡壳的 Redis 学习之路&#xff1a;从十大数据类型开始入手_AQin1012的博客-CSDN博客关于Redis的数据类型&#xff0c;各个文章总有些小不同&#xff0c;我们这里讨论的是Redis 7.0&#xff0c;为确保准确&#xff…
阅读更多...
Raft算法之Leader选举

Raft算法之Leader选举

Raft算法之Leader选举 一、Leader选举概述 Raft 使用心跳&#xff08;heartbeat&#xff09;触发Leader选举。当服务器启动时&#xff0c;初始化为Follower。Leader向所有Followers周期性发送heartbeat。如果Follower在选举超时时间内没有收到Leader的heartbeat&#xff0c;就…
阅读更多...
图像视频基础

图像视频基础

参考学习资料&#xff1a;https://blog.csdn.net/qq_28258885/article/details/116192244 文章目录 图像颜色深度分辨率 视频帧率比特率帧类型消除冗余的方法时间冗余&#xff08;帧间预测&#xff09;空间冗余&#xff08;帧内预测&#xff09; 视频编码器1.分区2.预测3.转换…
阅读更多...
软件测试基础教程学习4

软件测试基础教程学习4

文章目录 软件测试技术和方法4.1 静态测试和动态测试4.2 黑盒测试和白盒测试概述4.3 黑盒测试技术4.3.1 等价类划分4.3.2 边值分析4.3.3 因果图法4.3.4 正交实验设计法4.4.5 决策表驱动测试4.5.6 错误推荐法 4.4 白盒测试技术4.4.1 程序结构分析测试4.4.2 逻辑覆盖测试4.4.3 路…
阅读更多...
JSP页面跳转刷新

JSP页面跳转刷新

问题: 当前的jsp页面触发ajax请求后,能够获得新的相应页面,但是浏览器上展示的依然是老的页面,数据不刷新 尝试使用页面重定向依然无效, 最后使用js的window.location.href, 让浏览器的页面url 重加载才ok function submitDate() {var date1 document.getElementById("d…
阅读更多...
【uniapp】uniapp反向代理解决跨域问题（devServer）

【uniapp】uniapp反向代理解决跨域问题（devServer）

背景介绍 前段时间&#xff0c;在拿uniapp开发的时候&#xff0c;出现了跨域问题&#xff0c;按理说跨域应该由后端解决&#xff0c;但既然咱前端可以上&#xff0c;我想就上了&#xff08;顺手装个13&#xff09; 首先介绍什么是跨域 出于浏览器的同源策略&#xff0c;在发…
阅读更多...
docker-使用harbor搭建私有仓库

docker-使用harbor搭建私有仓库

前提 安装docker-ce 安装docker-compose 安装 安装docker-ce # step 1: 安装必要的一些系统工具 sudo yum install -y yum-utils device-mapper-persistent-data lvm2 # Step 2: 添加软件源信息 sudo yum-config-manager --add-repo https://mirrors.aliyun.com/docker-ce…
阅读更多...
金蝶云星空无需代码连接钉钉考勤的方法

金蝶云星空无需代码连接钉钉考勤的方法

金蝶云星空用户使用场景&#xff1a; 企业的销售渠道人员出差之前需要在金蝶云星空上提交出差申请单&#xff0c;并等待审批通过&#xff1b;每当销售任务完成&#xff0c;金蝶云星空上的审批通过后&#xff0c;需要人力在考勤系统中手动修改考勤信息。看似比较简单的流程&…
阅读更多...
基于Java+SpringBoot+vue的汽车改装方案网站设计与实现

基于Java+SpringBoot+vue的汽车改装方案网站设计与实现

博主介绍&#xff1a;✌擅长Java、微信小程序、Python、Android等&#xff0c;专注于Java技术领域和毕业项目实战✌ &#x1f345;文末获取源码联系&#x1f345; &#x1f447;&#x1f3fb; 精彩专栏推荐订阅&#x1f447;&#x1f3fb; 不然下次找不到哟 Java项目精品实战案…
阅读更多...
Bitbucket 新版本的安全限制

Bitbucket 新版本的安全限制

新版本的安全限制 是继续按照他给的第二个链接进入Bitbucket仓库后台添加App密码&#xff0c;也就是每个仓库需要单独的秘密码&#xff0c;这样的话就更加安全。 生产新密码&#xff1a; 这一坨务必要妥善保存&#xff0c;因为一旦点了关闭之后你就再也没有机会看到这个密码了…
阅读更多...
Hive一分钟

Hive一分钟

分区和分桶 1.分区表是将大的表文件划分成多个小文件以利于查询&#xff0c;但是如果数据分布不均衡&#xff0c;也会影响查询效率。 2.桶表可以对数据进行哈希取模&#xff0c;目的是让数据能够均匀的分布在表的各个文件中。 3.物理上&#xff0c;每个桶就是表和分区目录里的…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023