【SpringBoot系列】SpringBoot整合Kafka(含源码)

       在现代微服务架构的构建中，消息队列扮演着关键角色，码新而Apache Kafka凭借其高吞吐量、码新可扩展性和容错性脱颖而出。码新本文将深入讲解如何在SpringBoot框架中集成Kafka，码新以实现实时数据传输和处理。码新挂号源码

       Kafka是码新一个开源的流处理平台，由LinkedIn开发，码新专为大型实时数据流处理应用设计。码新它基于发布/订阅模式，码新支持分布式系统中的码新数据可靠传递，并可与Apache Storm、码新Hadoop、码新Spark等集成，码新应用于日志收集、码新大规模消息系统、用户活动跟踪、实时数据处理、指标聚合以及事件分发等场景。

       在集成SpringBoot和Kafka时，首先需要配置版本依赖。如果遇到如"Error connecting to node"的连接问题，可以尝试修改本地hosts文件，确保正确指定Kafka服务器的IP地址。成功整合后，成吉思汗端游源码SpringBoot将允许服务间高效地传递消息，避免消息丢失，极大地简化了开发过程。

       完整源码可通过关注公众号"架构殿堂"获取，回复"SpringBoot+Kafka"即可。最后，感谢您的支持和持续关注，"架构殿堂"公众号将不断更新AIGC、Java基础面试题、Netty、Spring Boot、Spring Cloud等实用内容，期待您的持续关注和学习。

kafka源码Topic的创建源码分析(附视频)

       关于Kafka Topic创建的源码分析，可以从kafka-topic.sh脚本的入口开始，它执行了kafka.admin.TopicCommand类。在创建Topic时，主要涉及AdminClientTopicService对象的创建和AdminClientClient创建Topics方法的调用，其中Controller负责处理客户端的CreateTopics请求。

       服务端的处理逻辑在KafkaRequestHandler.run()方法中，通过apis.handle(request)调用对应接口，如KafkaApis.handleCreateTopicsRequest，这个方法会触发adminManager.createTopics()，创建主题并监控其完成状态。制作官网源码创建的Topic配置和分区副本信息会被写入Zookeeper，如Topic配置和Topic的分区副本分配。

       当Controller监听到/brokers/topics/Topic名称的变更后，会触发Broker在磁盘上创建相关Log文件。如果Controller在创建过程中失败，如Controller挂掉，待重新选举后，创建过程会继续，直到Log文件被创建并同步到zk中。

       创建Topic时，zk上会创建特定节点，包括主题配置和分区信息。手动添加或删除/brokers/topics/节点将影响Topic的创建和管理。完整参数可通过sh bin/kafka-topic -help查看。

浅析源码 golang kafka sarama包(一)如何生产消息以及通过docker部署kafka集群with kraft

       本文将深入探讨Golang中使用sarama包进行Kafka消息生产的过程，以及如何通过Docker部署Kafka集群采用Kraft模式。首先，我们关注数据的生产部分。

       在部署Kafka集群时，我们将选择Kraft而非Zookeeper，通过docker-compose实现。集群中，理解LISTENERS的含义至关重要，主要有几个类型：

       Sarama在每个topic和partition下，web斗地主源码会为数据传输创建独立的goroutine。生产者操作的起点是创建简单生产者的方法，接着维护局部处理器并根据topic创建topicProducer。

       在newBrokerProducer中，run()方法和bridge的匿名函数是关键。它们反映了goroutine间的巧妙桥接，通过channel在不同线程间传递信息，体现了goroutine使用的精髓。

       真正发送消息的过程发生在AsyncProduce方法中，这是数据在三层协程中传输的环节，虽然深度适中，但需要仔细理解。

       sarama的架构清晰，但数据传输的核心操作隐藏在第三层goroutine中。输出变量的使用也有讲究：当output = p.bridge，它作为连接内外协程的桥梁；output = nil则关闭channel，output = bridge时允许写入。

flink 1. 1.区别

       å› ä¸ºåœ¨ Flink 1. 中，当前这类任务开发对于用户来说，还是不够友好，需要很多代码，同时也会造成 Flink SQL 冗长。

       Flink 1. SQL Connector 支持 Kafka Upsert Connector，这也是我们公司内部业务方对实时平台提出的需求。

收益：便利用户有这种需要从 kafka 取最新记录操作的实时任务开发，比如这种 binlog -> kafka，然后用户聚合操作，这种场景还是非常多的，这能提升实时作业开发效率，同时 1. 做了优化，性能会比单纯的 last_value 性能要好。

Flink Yarn 作业 On k8s 的生产级别能力是：

        Flink Jar 作业已经全部 K8s 化，Flink SQL 作业由于是推广初期，还是在 Yarn 上面进行运行，为了将实时计算 Flink 全部K8s化。

       æ‰€ä»¥æˆ‘们 Flink SQL 作业也需要迁移到 K8s，目前 Flink 1. 已经满足生产级别的 Flink k8s 功能，所以 Flink SQL K8s 化，打算直接使用社区的 On k8s 能力。

风险：虽然和社区的人沟通，Flink 1. on k8s 没有什么问题，但是具体功能还是需要先 POC 验证一下，同时可能社区 Flink on k8s 的能力。

       å¯èƒ½ä¼šé™åˆ¶æˆ‘们这边一些 k8s 功能使用，比如 hostpath volome 以及 Ingress 的使用，这里可能需要改底层源码来进行快速支持（社区有相关 JIRA 要做）。

Kafka Logcleaner源码分析

       Kafka日志保留策略包括按时间/大小和compact两种。Logcleaner遵循compact策略清理日志，只保留最新的消息，当多个消息具有相同key时，只保留最新的一个。

       每个日志由两部分组成：clean和dirty。dirty部分可以进一步划分为cleanable和uncleanable。偷菜游戏 源码uncleanable部分不允许清理，包括活跃段和未达到compact延迟时间的段。

       清理过程由后台线程定期执行，选择最脏的日志进行清理，脏度由dirty部分字节数与总字节数的比例决定。清理前，Logcleaner构建一个key->last_offset映射，包含dirty部分的所有消息。清理后，日志文件过滤掉过期消息，并合并较小的连续段为较大文件。

       payload为null的消息被Logcleaner删除，这类消息在topic配置的时间内保留，然后被清理。清理过程需与幂等性和事务性生产者兼容，保留活跃生产者最后一批消息，直到产生新消息或生产者不活跃。只清理提交或终止事物中的消息，未提交事物中的消息不清理。

       Logcleaner通过cleanOrSleep方法启动清理，选择最脏日志，调用clean清理并合并段。在清理前计算tombstone的移除时间，确保在clean部分驻留一定时间后移除。清理过程包括构建offset映射，分组段文件并清理合并。

       Logcleaner的清理逻辑确保了高效和一致的日志管理，助力Kafka系统稳定运行。

Kafka消费者源码：重平衡（1）-初始化与FIND_COORDINATOR

       在Kafka 2.5.2的消费者组中，重平衡是关键，它定义了消费者如何根据订阅关系调整对Topic分区的分配。当消费者数量、订阅的Topic或GroupCoordinator所在的Broker发生变更时，会触发重平衡。

       消费者组状态由GroupState类管理，共有五个状态：Empty（无成员）、PreparingRebalance（加入中）、CompletingRebalance（等待分配）、Stable（已平衡）和Dead（元数据已删除）。状态间的转换基于预先定义的前置状态。例如，从Empty到PreparingRebalance，预示着重平衡的开始。

       重平衡过程分为几个步骤，首先是消费者和Broker之间的协调。服务端启动时，GroupCoordinator组件即已就绪，而Consumer通过ConsumerCoordinator与之通信。在启动时，消费者首先会通过FindCoordinatorRequest找到GroupCoordinator，通过最小负载节点发送请求，然后服务端确定哪个Broker负责协调，如groupId的hash值对consumer_offsets分区数取模确定。

       一旦找到GroupCoordinator，消费者会发送JoinGroupRequest。后续步骤如SYNC_GROUP和HEARTBEAT确保消费者与协调器保持同步。这部分详细内容在后续的文章中会进一步探讨。

sarama 源码解析--Kafka的重平衡

       重平衡操作

       重平衡是动态调整Consumer Group下的Consumer订阅Topic的分区的一个关键操作。Sarama中的BalanceStrategyRange和BalanceStrategySticky策略具体实施这一操作。

       重平衡触发条件之一是成员数变更。这一过程包括以下步骤：

       1. 启动一个新的消费者实例。

       2. 调用Consume方法。

       3. Consume方法初始化连接信息，并启动一个goroutine。程序会阻塞在sess.ctx.Done()上。

       4. 在newSession方法中找到协调者信息，并发起join请求和syncgroup请求。Consumer Leader执行一次重平衡。

       5. 创建consumer group session，并初始化offset manager和开启心跳goroutine。

       6. 当心跳超时或收到coordinator的重平衡通知时，调用cancel()方法取消操作，退出Consume逻辑。

       7. 此时，Consume函数优雅退出。由于外层循环的存在，会重新执行Consume，实现一次重平衡。

       另一个触发重平衡的条件是订阅主题分区数发生变更。这一过程如下：

       1. 在Consume方法中开启心跳goroutine，并将consumer group session传递给它。

       2. 分区数发生变化时，调用sess.cancel()，Consume优雅退出并重新执行，实现重平衡。

Kafka源码分析(五) - Server端 - 基于时间轮的延时组件

       Kafka内部处理大量的延时操作，例如，在接收到PRODUCE请求后，副本可以等待一个timeout的时间再响应客户端。下面我们来探讨一个问题：为什么Kafka要自己实现一个延时任务组件，而不是直接使用Java的java.util.concurrent.DelayQueue呢？我们可以从以下两个方面来分析这个问题。

       1.1 DelayQueue的能力

       DelayQueue相关的接口/类如下所示：

       相应地，DelayQueue提供的能力如下：

       1.2 Kafka的业务场景

       Kafka的业务背景具有以下特点：

       相应地，Kafka对延时任务组件有以下两点要求：

       这两点要求都无法通过直接应用DelayQueue的方式得到满足。

       二. 组件接口

       让我们来看看Kafka的延时任务组件对外提供的接口，从而了解其提供的能力和使用方式。

       如下所示：

       左边的两个类定义了"延时操作"，右边的DelayedOperationPurgatory类定义了一个维护DelayOperaton的容器，其核心操作如下：

       三. 实现

       以下是关于"延时"实现方式的介绍。

       3.1 业务模型

       时间轮延时组件的思路如下：

       接下来，通过一个具体的例子来说明这种映射逻辑：

       首先关注上图中①号时间轮。圆环中的每一个单元格表示一个TimerTaskList。单元格有其关联的时间跨度；下方的"1s x "表示时间轮上共有个单元格，每个单元格的时间跨度为1秒。有一个指针指向了"当前时间"所对应的单元格。顺时针方向为时间流动方向。

       当收到一个延迟时间在0-1s的TimerTask时，会将其追加到①号时间轮的橙色单元格中。当收到一个延迟时间在3-4s的TimerTask时，会将其追加到①号时间轮的**单元格中。以此类推。

       现在有一个问题：①号时间轮能表示的最大延迟时间是秒，那如果收到了延迟秒的任务该怎么办？这时该用到②号时间轮了，我们称②号为①号的"溢出时间轮"。②号时间轮的特点如下：

       如此，延迟时间在-s的TimerTask会被追加到②号的紫色单元格，延迟时间在-s的TimerTask会被追加到②号的绿色单元格中。③号时间轮同理。

       刚刚是按①->②->③的顺序来分析时间轮的逻辑，反过来也可以得到有用的想象手里有一个"放大镜"，其实③号时间轮的蓝色单元格"放大"后是②号时间轮；②号时间轮的蓝色单元格"放大"后是①号时间轮；蓝色单元格并不实际存储TimerTask。

       3.2 数据结构

       DelayedOperationPurgatory有一个Timer类型的timeoutTimer属性，用于维护延时任务。实际使用的是Timer的实现类：SystemTimer。该类用于维护延时任务的核心属性有两个：delayQueue和timingWheel。TimingWheel表示单个时间轮，接下来我们来看看其类图：

       各属性含义如下：

       3.3 算法

       3.3.1 添加任务

       添加任务的入口是DelayedOperationPurgatory.tryCompleteElseWatch，其核心逻辑分为如下两步：

       SystemTimer.add直接调用了addTimerTaskEntry方法，后者逻辑如下：

       TimingWheel.add的逻辑也很清晰，分如下4种场景处理：

       3.3.2 尝试提前触发任务

       入口是DelayedOperationPurgatory.checkAndComplete：

       接下来看Watchers.tryCompleteWatched方法的内容：

       DelayedOperation.maybeTryComplete方法最终调用了DelayedOperation.tryComplete；

       DelayedOperation的子类需要在后者中实现自己的"触发条件"检查逻辑；若满足了提前触发的条件，则调用forceComplete方法执行事件触发场景下的业务逻辑。

       3.3.3 任务到期自动执行

       DelayedOperationPurgatory中维护了一个expirationReaper线程，其职责就是循环调用kafka.utils.timer.SystemTimer#advanceClock来从时间轮中获取已超时的任务，并更新时间轮的"当前时间"指针。

       四. 总结

       才疏学浅，未能窥其十之一二，随时欢迎各位交流补充。若文章质量还算及格，可以点赞收藏加以鼓励，后续我继续更新。

       另外，也可以在目录中找到同系列的其他文章：

       感谢阅读。