Skywalking Kafka Tracing实现
背景
Skywalking默认场景下,Tracing对于消息队列的发送场景,无法将TraceId传递到下游消费者,但对于微服务场景下,是有大量消息队列的业务场景的,这显然无法满足业务预期。
解决方案
Skywalking的官方社区中,有用户提出了该场景问题,Skywalking在补充工具包中,提供了对Kafka的tracing支持。
代码实现:
<dependency><groupId>org.apache.skywalking</groupId><artifactId>apm-toolkit-kafka</artifactId><version>${skywalking.version}</version></dependency>
对于该工具包,默认情况下,是针对KafkaTemplate进行trace,即如果使用KafkaTemplate发送消息,代码层面无需做任何改动。
如果没有使用KafkaTemplate的场景,toolkit也提供的了注解的支持:
public class ConsumerThread2 extends Thread {@Overridepublic void run() {Properties consumerProperties = new Properties();//...consumerProperties.put()KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);consumer.subscribe(topicPattern, new NoOpConsumerRebalanceListener());while (true) {if (pollAndInvoke(consumer)) break;}consumer.close();}@KafkaPollAndInvokeprivate boolean pollAndInvoke(KafkaConsumer<String, String> consumer) {try {Thread.sleep(1000);} catch (InterruptedException e) {}ConsumerRecords<String, String> records = consumer.poll(100);if (!records.isEmpty()) {OkHttpClient client = new OkHttpClient.Builder().build();Request request = new Request.Builder().url("http://localhost:8080/kafka-scenario/case/kafka-thread2-ping").build();Response response = null;try {response = client.newCall(request).execute();} catch (IOException e) {}response.body().close();return true;}return false;}
}
异步线程Tracing
对于Kafka消息的发送,经常会配合异步线程池的场景使用,Tracing的基本原理是基于ThreadLocal进行实现的,那么对于异步场景,是会丢失TraceId,通常的解决方式,是需要手动将主线程的TraceId手动赋值给子线程,但这种方式需要手动代码侵入,并不友好。
幸运的是,Skywalking的toolkit中提供了对于异步线程tracing的支持。
<dependency><groupId>org.apache.skywalking</groupId><artifactId>apm-toolkit-trace</artifactId><version>${skywalking.version}</version>
</dependency>
推荐用法:
ExecutorService executorService = Executors.newFixedThreadPool(1);
executorService.execute(RunnableWrapper.of(new Runnable() {@Override public void run() {//your code}
}));
或者:
@TraceCrossThreadpublic static class MyCallable<String> implements Callable<String> {@Overridepublic String call() throws Exception {return null;}}
...ExecutorService executorService = Executors.newFixedThreadPool(1);executorService.submit(new MyCallable());
PS:事实上,RunnableWrapper也是基于@TraceCrossThread实现。
相关文档:
https://skywalking.apache.org/docs/skywalking-java/v8.16.0/en/setup/service-agent/java-agent/application-toolkit-kafka/
https://skyapm.github.io/document-cn-translation-of-skywalking/zh/6.1.0/setup/service-agent/java-agent/Application-toolkit-trace-cross-thread.html
https://blog.51cto.com/knifeedge/5268667
https://blog.csdn.net/lijunwyf/article/details/107954543
相关文章:

Skywalking Kafka Tracing实现
背景 Skywalking默认场景下,Tracing对于消息队列的发送场景,无法将TraceId传递到下游消费者,但对于微服务场景下,是有大量消息队列的业务场景的,这显然无法满足业务预期。 解决方案 Skywalking的官方社区中…...
Perl 解析字符串为日期对象并获取多天前的日期字符串
Perl 解析字符串 perl语言中的库 Time::Piece可以将字符串解析为日期对象。 use Time::Piece; my ($y, $m, $d) 20230823 ~ /^([0-9]{4})([0-9]{2})([0-9]{2})\z/ or die;my $dt Time::Piece->strptime("$y$m$d","%Y%m%d");Perl获取多天之前的日期 …...
C语言问题 - 关于一维数组和二维数组用*a+i形式表达
问题 今天在编写程序时被一个语句搞懵了: #include<stdio.h>int main() {int *pa[6];int a[2][3] {{1,2,3},{4,5,6}};int sum 0;int i,j,k 0;for (i0; i<2; i){for (j0; j<3; j){*(pa k) *(a i) j;k;}}for (i0; i<k; i){printf("%d &q…...

验证码识别DLL ,滑块识别SDK,OCR图片转文字,机器视觉找物品
验证码识别DLL ,滑块识别SDK 你们用过哪些OCR提取文字,识图DLL,比如Opencv,Labview机器视觉找物品之类?...
【图论】最小生成树的应用
一.题目 P1550 [USACO08OCT] Watering Hole G - 洛谷 | 计算机科学教育新生态 (luogu.com.cn) 二.分析 1.我们是要使所有的农场都要有水 2.可以从起点引水,也可以互相引水。 3.费用要最小 这时我们可以想到最小生成树,建立一个虚拟节点即可。思路一…...
C++类模板的特化(三)
本文主要介绍类模板的特化、局部特化和缺省模板实参; 1.类模板的特化 类模板的特化(Class Template Specialization)是指为特定的模板参数提供自定义实现的过程。通过特化,我们可以针对某些特定的类型或条件提供不同的行为或实现…...

基于YOLOV8模型的课堂场景下人脸目标检测系统(PyTorch+Pyside6+YOLOv8模型)
摘要:基于YOLOV8模型的课堂场景下人脸目标检测系统可用于日常生活中检测与定位课堂场景下人脸,利用深度学习算法可实现图片、视频、摄像头等方式的目标检测,另外本系统还支持图片、视频等格式的结果可视化与结果导出。本系统采用YOLOv8目标检…...

java八股文面试[数据结构]——Map有哪些子类
知识来源: 【23版面试突击】 用过哪些Map类,都有什么区别,HashMap是线程安全的吗?_哔哩哔哩_bilibili https://www.cnblogs.com/bubbleboom/p/12694013.html...
司徒理财:8.23今日黄金原油走势分析附操作策略
黄金走势分析: 黄金下跌遇阻,短线开启震荡调整走势,但跌势依旧没有改变,没有突破1906压力前,还是偏空走势,反弹继续干空。趋势行情,不要轻言翻转!即便下跌结束,…...

使用动态IP是否会影响网络
今天我们要谈论的话题是关于动态IP和网络的关系。也许有些小伙伴对这个概念还比较陌生,但别担心,我会简单明了的给你理清楚。让我们一起看看动态IP到底能否影响到网络。 首先,我们先来搞明白什么是动态IP。在互联网世界中,每一个连…...
Linux学习笔记-常用指令说明
本文目录 一、Linux指令笔记 二、"授人以鱼,不如授人以渔" 一、Linux指令笔记 0、cd 命令是 change dir 的简写,它可以把终端当前所在的路径切换至目标路径。 1、mkdir 建立文件夹。是 make directory 的简写,它可以在文件系统中创建一个新的目…...
MyBatisPlus进阶版
1.映射 1.1自动映射 【1】表名和实体类名映射 -> 表名user 实体类名User 【2】字段名和实体类属性名映射 -> 字段名name 实体类属性名name 【3】字段名下划线命名方式和实体类属性小驼峰命名方式映射 -> 字段名 user_email 实体类属性名 userEmail MybatisPlus…...

安防视频云平台EasyNVR视频汇聚平台硬件无法进入服务器的问题处理方法
EasyNVR是基于RTSP/Onvif协议的视频接入、处理及分发的安防视频云平台,可提供的视频能力包括:设备接入、实时视频直播、录像、云存储、录像回放与检索、告警、级联等,平台可支持将接入的视频流进行全平台、全终端的分发,分发的视频…...

流媒体内容分发终极解决方案:当融合CDN与P2P视频交付结合
前言 随着互联网的发展,流媒体视频内容日趋增多,已经成为互联网信息的主要承载方式。相对传统的文字,图片等传统WEB应用,流媒体具有高数据量,高带宽、高访问量和高服务质量要求的特点,而现阶段互联网“尽力…...

根据源码,模拟实现 RabbitMQ - 内存数据管理(4)
目录 一、内存数据管理 1.1、需求分析 1.2、实现 MemoryDataCenter 类 1.2.1、ConcurrentHashMap 数据管理 1.2.2、封装交换机操作 1.2.3、封装队列操作 1.2.4、封装绑定操作 1.2.5、封装消息操作 1.2.6、封装未确认消息操作 1.2.7、封装恢复数据操作 一、内存数据管理…...
Apache Flume架构和原理
Apache Flume是一个开源的分布式、可靠的日志收集和聚合系统,旨在将大量的日志数据从不同的数据源(如应用程序、服务器、设备)收集到中心存储或数据湖中。Flume的架构设计允许用户在大规模数据流的情况下实现可靠的数据传输和处理。 Flume特性 Apache Flume是一个用于收集…...
代码随想录算法训练营day38 | LeetCode 509. 斐波那契数 70. 爬楼梯 746. 使用最小花费爬楼梯
509. 斐波那契数(题目链接:力扣(LeetCode)官网 - 全球极客挚爱的技术成长平台) 思路:经典的dp题。 int fib(int n){if(n 0 || n 1) return n;return fib(n-1) fib(n-2); } 70. 爬楼梯(题目…...

Linux基本指令【下】
欢迎来到Cefler的博客😁 🕌博客主页:那个传说中的man的主页 🏠个人专栏:题目解析 🌎推荐文章:题目大解析3 目录 👉🏻cat👉🏻echo(输出…...

向量检索:基于ResNet预训练模型构建以图搜图系统
1 项目背景介绍 以图搜图是一种向量检索技术,通过上传一张图像来搜索并找到与之相关的其他图像或相关信息。以图搜图技术提供了一种更直观、更高效的信息检索方式。这种技术应用场景和价值非常广泛,经常会用在商品检索及购物、动植物识别、食品识别、知…...

SpringBoot 响应头添加版本号、打包项目后缀添加版本号和时间
文章目录 响应头添加版本号获取版本号添加响应处理器请求结果 打包项目后缀添加版本号和时间实现打包结果 响应头添加版本号 获取版本号 在 pom.xml 中,在 project.version 下定义版本号 在 application.yml 获取 pom.xml 中 project.version 中的信息 添加响应处…...

AI-调查研究-01-正念冥想有用吗?对健康的影响及科学指南
点一下关注吧!!!非常感谢!!持续更新!!! 🚀 AI篇持续更新中!(长期更新) 目前2025年06月05日更新到: AI炼丹日志-28 - Aud…...

C++实现分布式网络通信框架RPC(3)--rpc调用端
目录 一、前言 二、UserServiceRpc_Stub 三、 CallMethod方法的重写 头文件 实现 四、rpc调用端的调用 实现 五、 google::protobuf::RpcController *controller 头文件 实现 六、总结 一、前言 在前边的文章中,我们已经大致实现了rpc服务端的各项功能代…...

从WWDC看苹果产品发展的规律
WWDC 是苹果公司一年一度面向全球开发者的盛会,其主题演讲展现了苹果在产品设计、技术路线、用户体验和生态系统构建上的核心理念与演进脉络。我们借助 ChatGPT Deep Research 工具,对过去十年 WWDC 主题演讲内容进行了系统化分析,形成了这份…...
【Linux】C语言执行shell指令
在C语言中执行Shell指令 在C语言中,有几种方法可以执行Shell指令: 1. 使用system()函数 这是最简单的方法,包含在stdlib.h头文件中: #include <stdlib.h>int main() {system("ls -l"); // 执行ls -l命令retu…...
QMC5883L的驱动
简介 本篇文章的代码已经上传到了github上面,开源代码 作为一个电子罗盘模块,我们可以通过I2C从中获取偏航角yaw,相对于六轴陀螺仪的yaw,qmc5883l几乎不会零飘并且成本较低。 参考资料 QMC5883L磁场传感器驱动 QMC5883L磁力计…...

UE5 学习系列(三)创建和移动物体
这篇博客是该系列的第三篇,是在之前两篇博客的基础上展开,主要介绍如何在操作界面中创建和拖动物体,这篇博客跟随的视频链接如下: B 站视频:s03-创建和移动物体 如果你不打算开之前的博客并且对UE5 比较熟的话按照以…...
系统设计 --- MongoDB亿级数据查询优化策略
系统设计 --- MongoDB亿级数据查询分表策略 背景Solution --- 分表 背景 使用audit log实现Audi Trail功能 Audit Trail范围: 六个月数据量: 每秒5-7条audi log,共计7千万 – 1亿条数据需要实现全文检索按照时间倒序因为license问题,不能使用ELK只能使用…...

以光量子为例,详解量子获取方式
光量子技术获取量子比特可在室温下进行。该方式有望通过与名为硅光子学(silicon photonics)的光波导(optical waveguide)芯片制造技术和光纤等光通信技术相结合来实现量子计算机。量子力学中,光既是波又是粒子。光子本…...

JVM虚拟机:内存结构、垃圾回收、性能优化
1、JVM虚拟机的简介 Java 虚拟机(Java Virtual Machine 简称:JVM)是运行所有 Java 程序的抽象计算机,是 Java 语言的运行环境,实现了 Java 程序的跨平台特性。JVM 屏蔽了与具体操作系统平台相关的信息,使得 Java 程序只需生成在 JVM 上运行的目标代码(字节码),就可以…...

CVE-2020-17519源码分析与漏洞复现(Flink 任意文件读取)
漏洞概览 漏洞名称:Apache Flink REST API 任意文件读取漏洞CVE编号:CVE-2020-17519CVSS评分:7.5影响版本:Apache Flink 1.11.0、1.11.1、1.11.2修复版本:≥ 1.11.3 或 ≥ 1.12.0漏洞类型:路径遍历&#x…...