rabbitmq整合skywalking并编写自定义插件增强

rabbitmq整合skywalking

首先先下载准备好skywalking 的服务端和ui控制台，java-agent

https://skywalking.apache.org/downloads/

整合skywalking

我的流程是在生产者和消费者服务中去引入一个mq的sdk，具体SDK的内容可以查看这篇文章

在sdk的pom文件中引入了skywalking依赖

<!-- skywalking -->
<dependency><groupId>org.apache.skywalking</groupId><artifactId>apm-toolkit-logback-1.x</artifactId><version>9.3.0</version>
</dependency>
<dependency><groupId>org.apache.skywalking</groupId><artifactId>apm-toolkit-trace</artifactId><version>9.3.0</version>
</dependency>

引入之后进行logback的配置，也可能你们是使用log4j这些，配置自行百度就行

生产者logback.xml配置

<?xml version="1.0" encoding="UTF-8"?>
<configuration><conversionRule conversionWord="tid" converterClass="org.apache.skywalking.apm.toolkit.log.logback.v1.x.LogbackPatternConverter"/><!-- 日志存放路径 --><property name="log.path" value="logs/sms-platform" /><!-- 日志输出格式 --><property name="log.pattern" value="%d{HH:mm:ss.SSS} [%tid] [%thread] %-5level %logger{20} - [%method,%line] - %msg%n" /><!-- 控制台输出 --><appender name="console" class="ch.qos.logback.core.ConsoleAppender"><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternLogbackLayout"><Pattern>${log.pattern}</Pattern></layout></encoder></appender><!-- 系统日志输出 --><appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-info.log</file><!-- 循环政策：基于时间创建日志文件 --><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 日志文件名格式 --><fileNamePattern>${log.path}/sys-info.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"><Pattern>${log.pattern}</Pattern></layout></encoder><filter class="ch.qos.logback.classic.filter.LevelFilter"><!-- 过滤的级别 --><level>INFO</level><!-- 匹配时的操作：接收（记录） --><onMatch>ACCEPT</onMatch><!-- 不匹配时的操作：拒绝（不记录） --><onMismatch>DENY</onMismatch></filter></appender><appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-error.log</file><!-- 循环政策：基于时间创建日志文件 --><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 日志文件名格式 --><fileNamePattern>${log.path}/sys-error.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"><Pattern>${log.pattern}</Pattern></layout></encoder><filter class="ch.qos.logback.classic.filter.LevelFilter"><!-- 过滤的级别 --><level>ERROR</level><!-- 匹配时的操作：接收（记录） --><onMatch>ACCEPT</onMatch><!-- 不匹配时的操作：拒绝（不记录） --><onMismatch>DENY</onMismatch></filter></appender><!-- 用户访问日志输出  --><appender name="sys-user" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-user.log</file><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 按天回滚 daily --><fileNamePattern>${log.path}/sys-user.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"><Pattern>${log.pattern}</Pattern></layout></encoder></appender><appender name="grpc" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender"><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"><Pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%X{tid}] [%thread] %-5level %logger{36} -%msg%n</Pattern></layout></encoder></appender><!-- 系统模块日志级别控制  --><logger name="com.aasee" level="info" /><!-- Spring日志级别控制  --><logger name="org.springframework" level="warn" /><root level="info"><appender-ref ref="console" /><appender-ref ref="grpc"/></root><!--系统操作日志--><root level="info"><appender-ref ref="file_info" /><appender-ref ref="file_error" /></root><!--系统用户操作日志--><logger name="sys-user" level="info"><appender-ref ref="sys-user"/></logger>
</configuration> 

消费者logback.xml配置

<?xml version="1.0" encoding="UTF-8"?>
<configuration><conversionRule conversionWord="tid" converterClass="org.apache.skywalking.apm.toolkit.log.logback.v1.x.LogbackPatternConverter"/><!-- 日志存放路径 --><property name="log.path" value="logs/sms-consumer" /><!-- 日志输出格式 --><property name="log.pattern" value="%d{HH:mm:ss.SSS} [%tid] [%thread] %-5level %logger{20} - [%method,%line] - %msg%n" /><!-- 控制台输出 --><appender name="console" class="ch.qos.logback.core.ConsoleAppender"><encoder><pattern>${log.pattern}</pattern></encoder></appender><appender name="grpc" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender"><encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder"><layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout"><Pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%X{tid}] [%thread] %-5level %logger{36} -%msg%n</Pattern></layout></encoder></appender><!-- 系统日志输出 --><appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-info.log</file><!-- 循环政策：基于时间创建日志文件 --><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 日志文件名格式 --><fileNamePattern>${log.path}/sys-info.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder><pattern>${log.pattern}</pattern></encoder><filter class="ch.qos.logback.classic.filter.LevelFilter"><!-- 过滤的级别 --><level>INFO</level><!-- 匹配时的操作：接收（记录） --><onMatch>ACCEPT</onMatch><!-- 不匹配时的操作：拒绝（不记录） --><onMismatch>DENY</onMismatch></filter></appender><appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-error.log</file><!-- 循环政策：基于时间创建日志文件 --><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 日志文件名格式 --><fileNamePattern>${log.path}/sys-error.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder><pattern>${log.pattern}</pattern></encoder><filter class="ch.qos.logback.classic.filter.LevelFilter"><!-- 过滤的级别 --><level>ERROR</level><!-- 匹配时的操作：接收（记录） --><onMatch>ACCEPT</onMatch><!-- 不匹配时的操作：拒绝（不记录） --><onMismatch>DENY</onMismatch></filter></appender><!-- 用户访问日志输出  --><appender name="sys-user" class="ch.qos.logback.core.rolling.RollingFileAppender"><file>${log.path}/sys-user.log</file><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><!-- 按天回滚 daily --><fileNamePattern>${log.path}/sys-user.%d{yyyy-MM-dd}.log</fileNamePattern><!-- 日志最大的历史 60天 --><maxHistory>60</maxHistory></rollingPolicy><encoder><pattern>${log.pattern}</pattern></encoder></appender><!-- 系统模块日志级别控制  --><logger name="com.aasee" level="info" /><!-- Spring日志级别控制  --><logger name="org.springframework" level="warn" /><root level="info"><appender-ref ref="console" /><appender-ref ref="grpc"/></root><!--系统操作日志--><root level="info"><appender-ref ref="file_info" /><appender-ref ref="file_error" /></root><!--系统用户操作日志--><logger name="sys-user" level="info"><appender-ref ref="sys-user"/></logger>
</configuration> 

发现问题

在完成了上述配置之后，启动项目，我发现消费者一直无法打印tid始终为n/a，经过一番查询后发现原来skywalking官方提供的消费插件不适合我们，其只针对官方原生Client实现扩展，但我们一般都是使用Spring Rabbitmq stater，所以我们无法实现链路追踪。@RabbitListener的工作原理可以看这篇文章https://blog.csdn.net/m0_72397750/article/details/140558744

总的来说就是我们一般使用注解来实现监听的，那么手动埋点就应该埋在开始处理消息的executeListener()方法这

将消息头复制回来

详细的代码我放在了https://github.com/Aas-ee/skywalking-apm-sniffer 注释很详细 里面还有skywalking-rabbitmq的官方插件代码可以用来对照学习，可以拉取后直接打成jar包放到skywalking-agent/plugin目录下使用（觉得有用的话麻烦点点star，这对我很重要），同时对于callback的打印也做了增强，但是不一定适用大家，主要用于学习思想和原理。