1.Spring Kafka：Retry Topic、源码DLT 的解析使用与原理
2.分布式链路追踪 SkyWalking 源码分析 —— DataCarrier 异步处理库
3.Flink Collector Output 接口源码解析
4.源码解析kafka删除topic
5.Metersphere 源码启动并做性能测试（一）

Spring Kafka：Retry Topic、DLT 的源码使用与原理

       Spring Kafka 在核心功能之外，扩展了Retry Topic和DLT（死信队列）的解析支持。这个增强在spring-kafka 2.7.及更高版本中可用，源码早期版本则不支持。解析电影源码

       默认情况下，源码当消费逻辑遇到异常，解析Spring Kafka会进行快速重试，源码最多次，解析每次无间隔。源码如果重试后依旧失败，解析它会尝试commit记录。源码重试的解析机制基于SeekUtils#doSeeks，可以通过自定义SeekToCurrentErrorHandler来调整，源码例如设置重试间隔和失败后将消息发送到DLT。

       定制SeekToCurrentErrorHandler后，异常后的处理会间隔秒重试3次，如果所有尝试都失败，消息会被转移到死信队列。这样的设计避免了单个消息重试占用消费线程，而是通过专用的retry线程处理。

       开启Retry Topic和DLT的使用可以通过注解和全局配置实现。@RetryableTopic注解可以应用在`@KafkaListener`方法上，设置默认重试3次，间隔1秒，如果重试后依然失败，消息将转到死信队列。用户还可以自定义死信处理逻辑。淘宝任务源码

       配置方面，可以调整重试次数、延迟时间和死信策略，支持Spring EL表达式。`fixedDelayTopicStrategy`的选择很重要，但具体策略可以根据需求调整。

       源码解析显示，Spring Kafka通过暂停和恢复分区实现延迟重试。消息在Retry Topic中带有延迟时间，监听器在消费前检查并暂停分区，确保在期望的时间重新开始消费。这种设计有助于控制消息的延迟时间。

       关于Retry Topic策略，FixedDelayStrategy有MULTIPLE_TOPICS和SINGLE_TOPIC两种。前者会创建多个主题以实现指数级增长的重试时间，而后者保持固定延迟，但可能在分区分配上产生不一致。如何配置多个retry线程，可以根据需要调整KafkaListener的并发设置或自定义ContainerFactory。

       对于更深入的学习和实践，可以参考GitHub上的Spring Kafka示例：github.com/TavenYin/tav...

分布式链路追踪 SkyWalking 源码分析 —— DataCarrier 异步处理库

       本文基于 SkyWalking 3.2.6 正式版，主要分享 SkyWalking Collector Remote 远程通信服务，用于 Collector 集群内部通信。Remote Module 应用于 SkyWalking 架构中，实现跨节点的流式处理。

       本文从接口到实现顺序解析 SkyWalking Collector Remote 的项目结构和组件，包括 RemoteModule、RemoteSenderService、源码漏洞检测RemoteClientService、RemoteClient、CommonRemoteDataRegisterService、RemoteDataRegisterService、RemoteDataIDGetter、RemoteDataInstanceCreatorGetter、RemoteSerializeService、RemoteDeserializeService。RemoteModule 实现 Module 抽象类，定义服务如 RemoteSenderService、RemoteDataRegisterService，创建 RemoteClient 实现远程通信。CommonRemoteDataRegisterService 用于注册数据类型对应的远程数据创建器和获取数据协议编号。

       接着，本文深入探讨基于 Google gRPC 的远程通信实现，包括 RemoteModuleGRPCProvider、GRPCRemoteSenderService、GRPCRemoteClientService、GRPCRemoteClient、RemoteCommonServiceHandler、GRPCRemoteSerializeService、GRPCRemoteDeserializeService。RemoteModuleGRPCProvider 提供基于 gRPC 的组件服务实现类，实现远程发送服务、客户端选择器和远程客户端服务。GRPCRemoteClient 实现基于 gRPC 的远程客户端，支持异步发送消息。

       最后，外卖网源码本文提及 SkyWalking Collector Remote 也支持基于 Kafka 的远程通信实现，但目前暂未完成。为了进一步学习 SkyWalking 的分布式链路追踪和远程通信机制，读者可以关注公众号芋道源码，获取 Java 源码解析、原理讲解、面试题、学习指南，回复「书籍」领取 Java 从入门到架构的 本书籍，加入技术群讨论 Java、后端、架构相关技术。

Flink Collector Output 接口源码解析

       Flink Collector Output 接口源码解析

       Flink中的Collector接口和其扩展Output接口在数据传递中起关键作用。Output接口增加了Watermark功能，是数据传输的基石。本文将深入解析collect方法及相关重要实现类，帮助理解数据传递的逻辑和场景划分。

       Collector和Output接口

       Collector接口有2个核心方法，Output接口则增加了4个功能，WatermarkGaugeExposingOutput接口则专注于显示Watermark值。主要关注collect方法，它是数据发送的核心操作，Flink中有多个Output实现类，针对不同场景如数据传递、Metrics统计、广播和时间戳处理。

       Output实现类分类

       Output类可以归类为：同一operatorChain内的数据传递（如ChainingOutput和CopyingChainingOutput）、跨operatorChain间（RecordWriterOutput）、酷乐源码统计Metrics（CountingOutput）、广播（BroadcastingOutputCollector）和时间戳处理（TimestampedCollector）。

       示例应用与调用链路

       通过一个示例，我们了解了Kafka Source与Map算子之间的数据传递使用ChainingOutput，而Map到Process之间的传递则用RecordWriterOutput。在不同Output的选择中，objectReuse配置起着决定性作用，影响性能和安全性。

       总结来说，ChainingOutput用于operatorChain内部，RecordWriterOutput处理跨chain，CountingOutput负责Metrics，BroadcastingOutputCollector用于广播，TimestampedCollector则用于设置时间戳。开启objectReuse会影响选择的Output类型。

       阅读推荐

       Flink任务实时监控

       Flink on yarn日志收集

       Kafka Connector更新

       自定义Kafka反序列化

       SQL JSON Format源码解析

       Yarn远程调试源码

       State Processor API状态操作

       侧流输出源码

       Broadcast流状态源码解析

       Flink启动流程分析

       Print SQL Connector取样功能

源码解析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删除指定的topic。

       若未配置`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完全清除的关键步骤。

Metersphere 源码启动并做性能测试（一）

       最近发现了一个开源测试平台——Metersphere，其在GitHub上广受好评。平台以Java语言编写，功能丰富，包括测试管理、接口测试、UI测试和性能测试。因此，我决定在本地尝试启动并进行性能测试。

       Metersphere的架构主要包括前端Vue和后端SpringBoot，数据库使用MySQL，缓存则依赖Redis。为了本地启动MS项目，首先需准备环境，参考其官方文档进行操作。在启动项目时，可能会遇到找不到特定类的错误，通常这是由于依赖问题导致的。解决这类问题，最常见的方式是注释掉相关的依赖和引用。如果遇到启动时出现依赖bean的问题，这可能是因为找不到对应的bean注入或调用方法时找不到对应的类。这种问题通常需要开发人员通过排查找到问题根源并解决，百度等资源是查找解决方案的有效途径。

       启动项目后，会观察到后台服务运行正常，接下来启动前端服务。执行`npm run serve`命令，如果项目已打包，这一步骤通常能成功启动前端。遇到前端加载失败的问题，可能需要重新打包项目，确保所有资源文件都能正常加载。

       接下来，进行性能测试的准备。Metersphere的性能测试流程包括发起压力测试、Node-controller拉起Jmeter执行测试、数据从Kafka流中获取并计算后存入MySQL数据库。在启动性能测试过程中，首先拉取Node-controller项目，需修改Jmeter路径，并确保本地环境支持Docker，因为Node-controller依赖Docker容器进行性能测试。Data-Streaming服务则负责解析Kafka数据并进行计算，需要确保Kafka服务已启动。

       启动Metersphere的backend和frontend后，配置压测资源池，添加本地Node-controller服务的地址和端口。性能测试分为通过JMX和引用接口自动化场景两种方式，可以模拟真实的网络请求。配置压力参数后，保存并执行性能测试，查看报告以了解测试结果。Metersphere的报告功能较为全面，值得深入研究。

       本地启动并执行性能测试的流程大致如上所述。在遇到问题时，查阅官方文档和利用百度等资源是解决问题的关键。Metersphere的官方文档提供了详尽的信息，对新用户来说是宝贵的学习资源。若仍有问题，可以考虑加入社区群寻求帮助。