共计 3198 个字符,预计需要花费 8 分钟才能阅读完成。
如何写一个 Skywalking trace 插件
javaagent
原理
美团技术团队 -Java 动静调试技术原理及实际
类图
实现
ConsumeMessageConcurrentlyInstrumentation
public class ConsumeMessageConcurrentlyInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
// 须要加强的类
private static final String ENHANCE_CLASS = "com.aliyun.openservices.shade.com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently";
// 须要加强的办法
private static final String CONSUMER_MESSAGE_METHOD = "consumeMessage";
// 减少的办法对应的拦截器
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.ons.v1.MessageConcurrentlyConsumeInterceptor";
// 结构器不须要拦挡
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {return new InstanceMethodsInterceptPoint[] {
// 新增一个拦截器
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
// 办法匹配
return named(CONSUMER_MESSAGE_METHOD);
}
@Override
public String getMethodsInterceptor() {return INTERCEPTOR_CLASS;}
@Override
public boolean isOverrideArgs() {return false;}
}
};
}
@Override
protected ClassMatch enhanceClass() {
// 须要加强的类
return HierarchyMatch.byHierarchyMatch(new String[] {ENHANCE_CLASS});
}
}
AbstractMessageConsumeInterceptor
public abstract class AbstractMessageConsumeInterceptor implements InstanceMethodsAroundInterceptor {
public static final String CONSUMER_OPERATION_NAME_PREFIX = "OnsRocketMQ/";
// 在办法前加强
@Override
public final void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
// 拿到办法参数,转换成音讯列表
List<MessageExt> msgs = (List<MessageExt>) allArguments[0];
// 从音讯中中获取 TraceId 等 Context 信息
ContextCarrier contextCarrier = getContextCarrierFromMessage(msgs.get(0));
// 创立一个 entry span
AbstractSpan span = ContextManager.createEntrySpan(CONSUMER_OPERATION_NAME_PREFIX + msgs.get(0)
.getTopic() + "/Consumer", contextCarrier);
span.setComponent(ComponentsDefine.ROCKET_MQ_CONSUMER);
SpanLayer.asMQ(span);
for (int i = 1; i < msgs.size(); i++) {ContextManager.extract(getContextCarrierFromMessage(msgs.get(i)));
}
}
// 异样解决
@Override
public final void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {ContextManager.activeSpan().log(t);
}
private ContextCarrier getContextCarrierFromMessage(MessageExt message) {ContextCarrier contextCarrier = new ContextCarrier();
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {next = next.next();
next.setHeadValue(message.getUserProperty(next.getHeadKey()));
}
return contextCarrier;
}
}
MessageConcurrentlyConsumeInterceptor
public class MessageConcurrentlyConsumeInterceptor extends AbstractMessageConsumeInterceptor {
// 在办法后处理
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
// 获取生产状态
ConsumeConcurrentlyStatus status = (ConsumeConcurrentlyStatus) ret;
if (status == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
// 生产状态为重试,则设置 span 呈现谬误
AbstractSpan activeSpan = ContextManager.activeSpan();
activeSpan.errorOccurred();
Tags.MQ_STATUS.set(activeSpan, status.name());
}
// 进行 span
ContextManager.stopSpan();
return ret;
}
}
我的项目:apm-ons-1.x-plugin
参考文档
- apm-ons-1.x-plugin
- 美团技术团队 -Java 动静调试技术原理及实际
分享并记录所学所见
正文完