1.Java函数式编程：Collector接口详解
2.opensips2.4源码分析模块的码详加载
3.Opentelemetry和Prometheus的remote-write-receiver的实验
4.分布式链路追踪 SkyWalking 源码分析 —— DataCarrier 异步处理库
5.Flink Collector Output 接口源码解析
6.Scroll源码解析

Java函数式编程：Collector接口详解

       在Java8中引入了函数式编程范式，使得开发人员能更直观地利用Stream进行操作，码详例如对列表或数组中的码详元素进行分组等。这一过程中，码详Collector接口发挥了关键作用，码详帮助实现复杂操作的码详图片同步源码简洁化。

       Collector接口的码详使用涵盖了三个泛型，具体功能则通过不同的码详方法实现。其核心在于提供从中间操作到最终结果的码详完整流程，包括生成容器、码详元素聚合以及结果合并等步骤。码详

       具体来看，码详如Collectors.toList()方法就是码详将元素收集到一个列表中。这个方法内部首先生成一个ArrayList作为容器，码详接着使用accumulator方法将元素添加至容器内，码详最后通过指定的执行方式完成操作。此过程中，方法的调用路径涉及多个类，最终实现将多个元素整合为一个有序列表。

       通过观察源码和运行过程，我们可以清晰地理解Collector接口的运作原理。例如，在实现自己的Collector接口时，需根据具体需求定义元素收集的逻辑，如去重操作。以Person对象的idCard字段去重为例，开发者可以设计特定的Collector实现类，来满足特定的业务需求。

       实现步骤主要包括描述需求、定义方法逻辑、验证结果等环节。这不仅有助于优化代码结构，还能提高开发效率和代码可读性。通过实现自定义的Collector接口，可以灵活地应对各种复杂场景，实现更加高效、简洁的代码编写。

opensips2.4源码分析模块的three 源码加载

       揭秘opensips 2.4源码中的模块加载奥秘

       在opensips 2.4的底层架构中，模块的加载过程由loadmodule指令主导，核心实现主要集中在sr_module.c的load_module函数上。这个函数是模块集成的关键，通过统一的接口<strong>struct module_exports</strong>对外展示，无论是静态模块如<strong>proto_udp.so</strong>和<strong>proto_tcp.so</strong>，还是动态模块，都遵循这一标准。

       动态模块加载的路径是由<strong>mpath_buf变量控制，作为sr_load_module参数的一部分，它默认设置在opensips安装路径下的<strong>opensips/lib/opensips/modules/</strong>。

       模块加载流程如下：

解析配置：loadmodule指令被整合到全局配置中，引导模块的初始化流程。

初始化模块：调用<strong>struct module_exports的函数指针，确保模块能够正确启动。

       理解模块的运作，关键在于它继承自<strong>struct module_exports，特别是其中的初始化函数<strong>preinit_f和<strong>init_f，它们是模块启动的核心步骤。

       在main.c中的<strong>init_modules函数中，这个流程被细致地执行：

       遍历所有模块，尝试执行<strong>preinit_f，可能出现失败但不影响后续步骤。

       调用<strong>init_f，设置init_done标志，标志着模块初始化完成。

       释放依赖信息，确保内存管理的完整性。

       在<strong>init_mod阶段，进一步执行以下操作：

       循环调用<strong>init_f

       统计模块数据，与全局的stats_collector紧密相连。

       注册管理接口到mi_cmds，以便于系统管理。

       模块函数的注册过程十分关键，通过<strong>struct module_exports中的cmds字段，与全局的modules结构体关联起来，通过find_export函数找到并调用相应的函数。

       值得注意的classpath源码是，为了避免命名冲突，模块函数的名称通常会加上前缀，以此来标识其特定的命名空间。

Opentelemetry和Prometheus的remote-write-receiver的实验

       实验目标：探索并实践Opentelemetry和Prometheus的集成，利用Prometheus的远程写功能与Opentelemetry的collector相结合，实现指标的主动推送，并通过Prometheus进行可视化管理。

       实验环境：需要准备一个运行的Prometheus实例，以及一个Opentelemetry的collector。具体配置和部署步骤需参照实验环境部分。

       实验过程：首先，配置Prometheus以抓取本地指标，通过修改Prometheus配置文件并启动windows_exporter实现本地指标的生成与输出。接着，配置和启动Opentelemetry的collector，确保其支持与Prometheus的远程写功能。在这一阶段，需要根据源代码（例如：wuqingtao/opentelemetry_demo/otel-collector-config.yaml）进行相应的调整。最后，通过执行指标生成命令（源代码来自：wuqingtao/opentelemetry_demo/app），确保指标能够被正确生成并主动推送至Prometheus。

       可视化面板：在Prometheus中设置抓取目标，通常为运行的Prometheus实例。配置完成后，访问Prometheus控制面板，通过采集器面板查看并管理指标。同时，利用Prometheus的可视化功能，对主动写入的指标进行分析与监控。

       实验结果：借助Prometheus的远程写功能和Opentelemetry的collector，实现了指标的主动推送至Prometheus。这一集成使得实时监控和分析数据成为可能，进一步强化了监控系统的能力，提升了数据处理效率。

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

       本文基于 SkyWalking 3.2.6 正式版，主要分享 SkyWalking Collector Remote 远程通信服务，用于 Collector 集群内部通信。43源码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 的分布式链路追踪和远程通信机制，读者可以关注公众号芋道源码，ride 源码获取 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取样功能

Scroll源码解析

       1. Scroll查询在指定_doc排序时相较于不指定排序或指定某个字段排序能明显更快，这是由于Scroll查询的机制及底层实现所致。

        首先查看Elasticsearch的Collector，其主要功能是收集文档并按照特定规则排序。其中，TopDocsCollector类在收集文档后会返回一个有序的TopDocs对象，该对象是搜索结果的返回值。TopDocsCollector有三个子类：SimpleFieldCollector、PagingFieldCollector、SimpleTopScoreDocCollector 和PagingTopScoreDocCollector。这些子类根据排序规则（如字段排序、简单排序等）进行文档排序。

       2. 对于TopScoreDocCollector，其排序规则是先执行打分，分数相同的文档按文档号排序。TopFieldCollector则是先按照指定字段排序，值相同的文档再按文档号排序。

       3. TopScoreDocsCollector的两个子类（SimpleTopScoreDocCollector和PagingTopScoreDocCollector）在功能上区别在于PagingTopScoreDocCollector针对翻页请求，代码上增加了对after的判断。对于使用TopScoreDocsCollector无论是否为翻页请求，每次请求都会扫描全部命中文档并计算分值。使用SimpleTopScoreDocCollector还是PagingTopScoreDocCollector取决于after是否为null。

       4. 对于scroll请求，after参数等于scrollContext.lastEmittedDoc，即上次翻页最大的ScoreDoc。TopFieldCollector同样有两个子类（SimpleFieldCollector和PagingFieldCollector），其判断逻辑与TopScoreDocsCollector类似，也是根据searchContext.sort()是否为null来决定使用哪类Collector。

       5. 在lucene6.4.1版本中，无论是SimpleFieldCollector和PagingFieldCollector都无法提前终止收集过程。然而，从更高版本的lucene开始，具备了提前结束收集的功能，判断依据是search sort=index sort一致时，通过抛出CollectionTerminatedException异常提前结束收集。Elasticsearch从6.x版本开始也支持了自定义写入顺序，可以不是_doc而是某个字段值。

       6. 通过Elasticsearch的代码分析，我们确认scroll请求在指定_doc排序并从第二页开始时，只会收集指定数量的doc，性能表现更优。对于scroll请求，包装了一层MinDocQuery，用于过滤掉已经翻页过的数据，大大减少文档命中数，避免收集无用的doc，这对于深度翻页性能提升明显。

       7. 对于scroll请求，由于不支持向前翻页，每次查询对于已查过的数据无需收集。Elasticsearch通过MinDocQuery实现跳跃功能，将doc跳到segmentMinDoc（lastEmittedDoc+1），在合并倒排表之后，实际上就不会再命中上一页的内容。触发提前终止后，后续倒排表合并也不再必要，性能提升显著。

       8. Scroll与search_after查询实际上走的是相同的逻辑，都是通过一个after变量进行翻页。scroll的after参数为scrollContext.lastEmittedDoc（ScoreDoc），search_after的after参数为包含sort字段信息的FieldDoc，都是ScoreDoc。最终都会收集全部命中文档才能得到排序结果，但scroll对于_doc排序做了优化，性能表现更佳。

       9. 对于search_after查询，即使指定_doc排序，仍然需要收集全部命中文档，因为search_after是动态的，MinDocQuery跳跃功能不适用。然而，search_after在lucene后续版本中支持了提前终止功能，当查询时指定sort为index sort，可以触发提前终止，不再收集全部命中文档。

       . Scroll请求保存的上下文信息主要是maxScore和lastEmittedDoc用于翻页，但实际保存的不仅仅是ScrollContext，而是SearchContext，其中包含了更多关键信息，如searcher和IndexReader，后者对于后续索引更新是感知不到的，除非重新打开reader或使用DirectoryReader.openIfChanged(oldreader)。这是Scroll查询无法感知索引更新的原因。

       . 经过测试，即使在scroll过程中触发了merge，被merge的segment文件也不会立即被删除，新的segment文件也不会被发现。这表明Scroll查询无法感知数据更新，其本质是快照了LeafReaderContext，并非检索命中的结果。

       总结而言，Scroll查询在指定_doc排序时，通过优化收集过程和使用MinDocQuery实现跳跃功能，能显著提升性能，尤其是在翻页操作中。同时，Scroll请求的机制及底层实现使得其在查询处理上与search_after查询存在显著差异，但在Elasticsearch6.x版本中引入了索引预排序和提前终止功能，进一步优化了查询性能。

[UVM源代码研究] 谈谈uvm中的浅拷贝（shallow copy）与深拷贝（deep copy）

       在探讨UVM（Universal Verification Methodology）中的浅拷贝（shallow copy）与深拷贝（deep copy）之前，我们先对相关概念进行简要介绍，以便于理解以下讨论。浅拷贝和深拷贝是对象编程领域中基本概念，不仅限于系统Verilog（SV）和UVM（Universal Verification Methodology）。

       浅拷贝：这一概念涉及的是拷贝对象的指针，即浅拷贝只复制指向对象内存空间的指针，使得目标对象与源对象共享同一内存空间。浅拷贝的局限性在于当内存空间被销毁时，所有指向该空间的指针必须重新定义，否则会导致野指针错误。

       深拷贝：与此相反，深拷贝确保源对象和拷贝对象完全独立，两者之间互不影响，包括内存空间内容也被复制一份。例如，基本类型如Int、Double，以及结构体（struct）、枚举（Enum）会自动执行深拷贝，而类类型的对象则需区分浅拷贝与深拷贝。

       在UVM中，`uvm_object`类提供了`copy`与`clone`函数来实现对象的拷贝。

       `copy`函数为非虚拟、无返回值的函数，不能被重写，但`do_copy`函数为虚拟函数，可以通过重写`do_copy`函数实现对`copy`函数的间接重写。调用`copy`函数前，目标对象需先创建，以实现源对象内部对象的深拷贝赋值，而不会对目标对象本身分配空间。

       `clone`函数为虚拟函数，返回`uvm_object`类型，可以被重写。由于返回值类型限制，`clone`只能通过`$cast`来实现目标对象类型的转换，而不能直接赋值。`clone`函数返回一个指向源对象类型的`uvm_object`句柄，因此目标对象类型必须与源对象一致（通过`$cast`检查），以确保成功执行`clone`操作，且目标对象不需要事先分配空间，因为`clone`会自动分配新空间。

       `copy`函数的实现中，除了`do_copy`之外的第行的`__m_uvm_field_automation(rhs, UVM_COPY, "")`完成了在`field_automation`中的配置实现。如果未重写`do_copy`函数，则所有拷贝行为依赖于`__m_uvm_field_automation`函数。

       `uvm_object_defines.svh`文件在第行实现了将`copy`传入参数转换为局部变量`local_data__`，该变量类型为通过`uvm_object_untils_begin`传入的参数类型。`local_data__`在后续的`uvm_field_automation`宏中根据传入的标志位进行相应操作，以`uvm_field_object`为例。

       在`uvm_field_object`中，关于`UVM_COPY`的具体操作表明，调用`copy`的源对象不能为空。如果`FLAG&UVM_NOCOPY`位为1，则直接结束代码执行。如果`FLAG&UVM_REFERENCE`位为1，或者`local_data__.ARG == null`，则将目标对象的`ARG`对象句柄指向源对象的`ARG`句柄。这种做法对于未分配空间的对象赋值，以避免错误。`UVM_REFERENCE`的应用场景主要针对`uvm_component`类型的对象注册，确保在进行`copy`和`clone`时执行浅拷贝，避免深拷贝导致的问题。

       `uvm_component`类型在`copy`时默认执行深拷贝，而`UVM_REFERENCE`标志位则实现浅拷贝。例如，在`apb_env`中，`bus_monitor`和`bus_collector`被例化为`master`中的`monitor`和`collector`，同时`cfg`对象也传递给`master`。通过`field_automation`的修改，可以观察到`uvm_top`在打印树型结构时，`apb_monitor`和`cfg`对象的打印信息。

       总结而言，UVM中的默认拷贝/克隆操作为深拷贝，`UVM_REFERENCE`标志位用于实现浅拷贝。理解这些概念对于在UVM中进行对象拷贝时避免错误至关重要。

OpenTelemetry、Spring Cloud Sleuth、Kafka、Jager实现分布式跟踪

        分布式跟踪可让您深入了解特定服务在分布式软件系统中作为整体的一部分是如何执行的。它跟踪和记录从起点到目的地的请求以及它们经过的系统。

       

        在本文中，我们将使用 OpenTelemetry、Spring Cloud Sleuth、Kafka 和 Jaeger 在三个 Spring Boot 微服务 中实现分布式跟踪。

       

        我们先来看看分布式追踪中的一些基本术语。

       

        跨度：表示系统内的单个工作单元。跨度可以相互嵌套以模拟工作的分解。例如，一个跨度可能正在调用一个 REST 端点，然后另一个子跨度可能是该端点调用另一个，等等在不同的服务中。

       

        Trace：所有共享相同根跨度的跨度集合，或者更简单地说，将所有跨度创建为原始请求的直接结果。跨度的层次结构（每个跨度在根跨度旁边都有自己的父跨度）可用于形成有向无环图，显示请求在通过各种组件时的路径。

       

       

        OpenTelemetry ，也简称为 OTel，是一个供应商中立的开源 Observability 框架，用于检测、生成、收集和导出遥测数据，例如 跟踪 、 指标 和 日志 。作为 云原生 计算基金会 (CNCF) 的孵化项目，OTel 旨在提供与供应商无关的统一库和 API 集——主要用于收集数据并将其传输到某处。OTel 正在成为生成和管理遥测数据的世界标准，并被广泛采用。

       

       

        Sleuth 是一个由 Spring Cloud 团队管理和维护的项目，旨在将分布式跟踪功能集成到 Spring Boot 应用程序中。它作为一个典型Spring Starter的 . 以下是一些开箱即用的 Sleuth 工具：

       

       

        Sleuth 添加了一个拦截器，以确保在请求中传递所有跟踪信息。每次调用时，都会创建一个新的 Span。它在收到响应后关闭。

       

        Sleuth 能够跟踪您的请求和消息，以便您可以将该通信与相应的日志条目相关联。您还可以将跟踪信息导出到外部系统以可视化延迟。

       

       

        Jaeger 最初由 Uber 的团队构建，然后于 年开源。它于 年被接受为云原生孵化项目，并于 年毕业。作为 CNCF 的一部分，Jaeger 是云原生 架构 中公认的项目。它的源代码主要是用 Go 编写的。Jaeger 的架构包括：

       

       

        与 Jaeger 类似，Zipkin 在其架构中也提供了相同的组件集。尽管 Zipkin 是一个较老的项目，但 Jaeger 具有更现代和可扩展的设计。对于此示例，我们选择 Jaeger 作为后端。

       

       

        让我们设计三个 Spring Boot 微服务：

       

       

       

       

       

        这三个微服务旨在：

       

       

        这是为了观察 OpenTelemetry 如何结合 Spring Cloud Sleuth 处理代码的自动检测以及生成和传输跟踪数据。上面的虚线捕获了微服务导出的跟踪数据的路径，通过OTLP（OpenTelemetry Protocol）传输到OpenTelemetry Collector，收集器依次处理并将跟踪数据导出到后端Jaeger进行存储和查询。

       

        使用 monorepo，我们的项目结构如下：

       

       

       

       

        第 1 步：添加 POM 依赖项

       

        这是使用 OTel 和 Spring Cloud Sleuth 实现分布式跟踪的关键。我们的目标是不必手动检测我们的代码，因此我们依靠这些依赖项来完成它们设计的工作——自动检测我们的代码，除了跟踪实现、将遥测数据导出到 OTel 收集器等。

       

       

       

        第 2 步：OpenTelemetry 配置

       

        OpenTelemetry 收集器端点

       

        对于每个微服务，我们需要在其中添加以下配置application.yml（请参阅下面部分中的示例片段）。spring.sleuth.otel.exporter.otlp.endpoint主要是配置OTel Collector端点。它告诉导出器，在我们的例子中是 Sleuth，通过 OTLP 将跟踪数据发送到指定的收集器端点pose 服务。

       

        跟踪数据概率抽样

       

        spring.sleuth.otel.config.trace-id-ratio-based属性定义了跟踪数据的采样概率。它根据提供给采样器的分数对一部分迹线进行采样。概率抽样允许 OpenTelemetry 跟踪用户通过使用随机抽样技术降低跨度收集成本。如果该比率小于 1.0，则某些迹线将不会被导出。对于此示例，我们将采样配置为 1.0、%。

       

        有关其他 OTel Spring Cloud Sleuth 属性，请参阅常见应用程序属性。

       

       

        OpenTelemetry 配置文件

       

        我们需要项目根目录下的 OTel 配置文件otel-config.yaml。内容如下。此配置文件定义了 OTel 接收器、处理器和导出器的行为。正如我们所看到的，我们定义了我们的接收器来监听 gRPC 和 HTTP，处理器使用批处理和导出器作为 jaeger 和日志记录。

       

       

        第 3 步：docker-compose 将所有内容串在一起

       

        让我们看看我们需要启动哪些 docker 容器来运行这三个微服务并观察它们的分布式跟踪，前三个微服务在上面的部分中进行了解释。

       

       

       

        运行docker-compose up -d以调出所有九个容器：

       

       

       

        第 4 步：追踪数据在行动

       

        快乐之路

       

        现在，让我们启动customer-service-bff流程的入口点，以创建新客户。

       

       

       

        启动 Jaeger UI， /?target=http%3A//localhost%3A/%2C]按[/url]服务搜索customer-service-bff，单击Find Traces按钮，这是我们看到的创建客户跟踪：它跨越三个服务，总共跨越六个，持续时间 . 毫秒。

       

       

       

        除了 Trace Timeline 视图（上面的屏幕截图），Jaeger 还提供了一个图形视图（Trace Graph在右上角的下拉菜单中选择）：

       

       

       

        三个微服务在 docker 中的日志输出显示相同的跟踪 id，以红色突出显示，并根据其应用程序名称显示不同的跨度 id（应用程序名称及其对应的跨度 id 以匹配的颜色突出显示）。在 的情况下customer-service，相同的 span id 从 REST API 请求传递到 Kafka 发布者请求。

       

       

       

       

       

       

       

       

        customer-service让我们在 docker 中暂停我们的PostgreSQL 数据库，然后重复从customer-service-bff. internal server error正如预期的那样，我们得到了。检查 Jaeger，我们看到以下跟踪，异常堆栈跟踪抱怨SocketTimeoutException，再次如预期的那样。

       

       

       

        识别长期运行的跨度

       

        Jaeger UI 允许我们搜索超过指定最大持续时间的跟踪。例如，我们可以搜索所有耗时超过 毫秒的跟踪。然后，我们可以深入研究长期运行的跟踪以调查其根本原因。

       

       

        在这个故事中，我们从 OpenTelemetry、Spring Cloud Sleuth 和 Jaeger 的角度解压了分布式跟踪，验证了 REST API 调用和 Kafka pub/sub 中分布式跟踪的自动检测。我希望这个故事能让你更好地理解这些跟踪框架和工具，尤其是 OpenTelemetry，以及它如何从根本上改变我们在 分布式系统 中进行可观察性的方式。