1.sarama 源码解析--Kafka的消消费重平衡
2.Kafka Logcleaner源码分析
3.源码解析kafka删除topic
4.浅析源码 golang kafka sarama包(一)如何生产消息以及通过docker部署kafka集群with kraft
sarama 源码解析--Kafka的重平衡
重平衡操作
重平衡是动态调整Consumer Group下的Consumer订阅Topic的分区的一个关键操作。Sarama中的费源BalanceStrategyRange和BalanceStrategySticky策略具体实施这一操作。
重平衡触发条件之一是码k码成员数变更。这一过程包括以下步骤：
1. 启动一个新的消消费消费者实例。
2. 调用Consume方法。费源
3. Consume方法初始化连接信息，码k码求110的源码并启动一个goroutine。消消费程序会阻塞在sess.ctx.Done()上。费源
4. 在newSession方法中找到协调者信息，码k码并发起join请求和syncgroup请求。消消费Consumer Leader执行一次重平衡。费源
5. 创建consumer group session，码k码并初始化offset manager和开启心跳goroutine。消消费
6. 当心跳超时或收到coordinator的费源重平衡通知时，调用cancel()方法取消操作，码k码退出Consume逻辑。
7. 此时，Consume函数优雅退出。由于外层循环的直播app源码php存在，会重新执行Consume，实现一次重平衡。
另一个触发重平衡的条件是订阅主题分区数发生变更。这一过程如下：
1. 在Consume方法中开启心跳goroutine，并将consumer group session传递给它。
2. 分区数发生变化时，调用sess.cancel()，Consume优雅退出并重新执行，实现重平衡。
Kafka Logcleaner源码分析
Kafka日志保留策略包括按时间/大小和compact两种。Logcleaner遵循compact策略清理日志，只保留最新的消息，当多个消息具有相同key时，只保留最新的一个。
每个日志由两部分组成：clean和dirty。dirty部分可以进一步划分为cleanable和uncleanable。uncleanable部分不允许清理，包括活跃段和未达到compact延迟时间的app 视频解析 源码段。
清理过程由后台线程定期执行，选择最脏的日志进行清理，脏度由dirty部分字节数与总字节数的比例决定。清理前，Logcleaner构建一个key->last_offset映射，包含dirty部分的所有消息。清理后，日志文件过滤掉过期消息，并合并较小的连续段为较大文件。
payload为null的消息被Logcleaner删除，这类消息在topic配置的时间内保留，然后被清理。清理过程需与幂等性和事务性生产者兼容，保留活跃生产者最后一批消息，直到产生新消息或生产者不活跃。只清理提交或终止事物中的消息，未提交事物中的消息不清理。
Logcleaner通过cleanOrSleep方法启动清理，棋牌php源码下载选择最脏日志，调用clean清理并合并段。在清理前计算tombstone的移除时间，确保在clean部分驻留一定时间后移除。清理过程包括构建offset映射，分组段文件并清理合并。
Logcleaner的清理逻辑确保了高效和一致的日志管理，助力Kafka系统稳定运行。
源码解析kafka删除topic
本文以kafka0.8.2.2为例，解析如何删除一个topic以及其背后的关键技术和源码实现过程。
删除一个topic涉及两个关键点：配置删除参数以及执行删除操作。
首先，配置参数`delete.topic.enable`为`True`，这是Broker级别的配置，用于指示kafka是否允许执行topic删除操作。
其次，执行命令`bin/kafka-topics.sh --zookeeper zk_host:port/chroot --delete --topic my_topic_name`，此命令指示kafka删除指定的android 系统源码 idetopic。
若未配置`delete.topic.enable`为`True`，topic仅被标记为删除状态，而非立即清除。此时，通常的做法是手动删除Zookeeper中的topic信息和日志，但这仅会清除Zookeeper的数据，并不会真正清除kafkaBroker内存中的topic数据。因此，最佳做法是配置`delete.topic.enable`为`True`，然后重启kafka。
接下来，我们介绍几个关键类和它们在删除topic过程中的作用。
1. **PartitionStateMachine**：该类代表分区的状态机，决定分区的当前状态及其转移。状态包括：NonExistentPartition、NewPartition、OnlinePartition、OfflinePartition。
2. **ReplicaManager**：负责管理当前机器的所有副本，处理读写、删除等具体操作。读写操作流程包括获取partition对象，再获取Replica对象，接着获取Log对象，并通过其管理的Segment对象将数据写入、读出。
3. **ReplicaStateMachine**：副本的状态机，决定副本的当前状态和状态之间的转移。状态包括：NewReplica、OnlineReplica、OfflineReplica、ReplicaDeletionStarted、ReplicaDeletionSuccessful、ReplicaDeletionIneligible、NonExistentReplica。
4. **TopicDeletionManager**：管理topic删除的状态机，包括发布删除命令、监听并开始删除topic、以及执行删除操作。
在删除topic的过程中，分为四个阶段：客户端执行删除命令、未配置`delete.topic.enable`的流水、配置了`delete.topic.enable`的流水、以及手动删除Zookeeper上topic信息和磁盘数据。
客户端执行删除命令时，会在"/admin/delete_topics"目录下创建topicName节点。
未配置`delete.topic.enable`时，topic删除流程涉及监听topic删除命令、判断`delete.topic.enable`状态、标记topic为不可删除、以及队列删除topic任务。
配置了`delete.topic.enable`时，额外步骤包括停止删除topic、检查特定条件、更新删除topic集合、激活删除线程、执行删除操作，如解除分区变动监听、清除内存数据结构、删除副本数据、删除Zookeeper节点信息等。
关于手动删除Zookeeper上topic信息和磁盘数据，通常做法是删除Zookeeper的topic相关信息及磁盘数据，但这可能导致部分内存数据未清除。是否会有隐患，需要进一步测试。
总结而言，kafka的topic删除流程基于Zookeeper实现，通过配置参数、执行命令、管理状态机以及清理相关数据，以实现topic的有序删除。正确配置`delete.topic.enable`并执行删除操作是确保topic完全清除的关键步骤。
浅析源码 golang kafka sarama包(一)如何生产消息以及通过docker部署kafka集群with kraft
本文将深入探讨Golang中使用sarama包进行Kafka消息生产的过程，以及如何通过Docker部署Kafka集群采用Kraft模式。首先，我们关注数据的生产部分。
在部署Kafka集群时，我们将选择Kraft而非Zookeeper，通过docker-compose实现。集群中，理解LISTENERS的含义至关重要，主要有几个类型：
Sarama在每个topic和partition下，会为数据传输创建独立的goroutine。生产者操作的起点是创建简单生产者的方法，接着维护局部处理器并根据topic创建topicProducer。
在newBrokerProducer中，run()方法和bridge的匿名函数是关键。它们反映了goroutine间的巧妙桥接，通过channel在不同线程间传递信息，体现了goroutine使用的精髓。
真正发送消息的过程发生在AsyncProduce方法中，这是数据在三层协程中传输的环节，虽然深度适中，但需要仔细理解。
sarama的架构清晰，但数据传输的核心操作隐藏在第三层goroutine中。输出变量的使用也有讲究：当output = p.bridge，它作为连接内外协程的桥梁；output = nil则关闭channel，output = bridge时允许写入。