聊聊skywalking的rocketmq-plugin

我就是我 2023-09-24 00:29 26阅读 0赞

### 序 ###

本文主要研究一下skywalking的rocketmq-plugin

### skywalking-plugin.def ###

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/resources/skywalking-plugin.def

rocketMQ-4.x=org.apache.skywalking.apm.plugin.rocketMQ.v4.define.ConsumeMessageConcurrentlyInstrumentation
    rocketMQ-4.x=org.apache.skywalking.apm.plugin.rocketMQ.v4.define.ConsumeMessageOrderlyInstrumentation
    rocketMQ-4.x=org.apache.skywalking.apm.plugin.rocketMQ.v4.define.MQClientAPIImplInstrumentation
    rocketMQ-4.x=org.apache.skywalking.apm.plugin.rocketMQ.v4.define.SendCallbackInstrumentation
    复制代码

*  skywalking的rocketmq-plugin定义了4个增强，分别是ConsumeMessageConcurrentlyInstrumentation、ConsumeMessageOrderlyInstrumentation、MQClientAPIImplInstrumentation、SendCallbackInstrumentation

### ConsumeMessageConcurrentlyInstrumentation ###

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java

publicclassConsumeMessageConcurrentlyInstrumentationextendsClassInstanceMethodsEnhancePluginDefine {
        privatestatic final StringENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently";
        privatestatic final StringCONSUMER_MESSAGE_METHOD = "consumeMessage";
        privatestatic final StringINTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageConcurrentlyConsumeInterceptor";
    
        @OverridepublicConstructorInterceptPoint[] getConstructorsInterceptPoints() {
            returnnewConstructorInterceptPoint[0];
        }
    
        @OverridepublicInstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
            returnnewInstanceMethodsInterceptPoint[] {
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(CONSUMER_MESSAGE_METHOD);
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnINTERCEPTOR_CLASS;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                }
            };
        }
    
        @OverrideprotectedClassMatchenhanceClass() {
            returnHierarchyMatch.byHierarchyMatch(newString[] {ENHANCE_CLASS});
        }
    }
    复制代码

*  ConsumeMessageConcurrentlyInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageConcurrentlyConsumeInterceptor增强org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently的consumeMessage方法

#### AbstractMessageConsumeInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/AbstractMessageConsumeInterceptor.java

public abstract class AbstractMessageConsumeInterceptor implements InstanceMethodsAroundInterceptor {
    
        public static final String CONSUMER_OPERATION_NAME_PREFIX = "RocketMQ/";
    
        @Override
        public final void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
            Class<?>[] argumentsTypes,
            MethodInterceptResult result) throws Throwable {
            List<MessageExt> msgs = (List<MessageExt>)allArguments[0];
    
            ContextCarrier contextCarrier = getContextCarrierFromMessage(msgs.get(0));
            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().errorOccurred().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;
        }
    }
    复制代码

*  AbstractMessageConsumeInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法执行ContextManager.extract(getContextCarrierFromMessage(msgs.get(i)))；其handleMethodException方法执行ContextManager.activeSpan().errorOccurred().log(t)

#### MessageConcurrentlyConsumeInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/MessageConcurrentlyConsumeInterceptor.java

public classMessageConcurrentlyConsumeInterceptorextendsAbstractMessageConsumeInterceptor{
    
        @Override
        public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret) throwsThrowable {
            ConsumeConcurrentlyStatus status = (ConsumeConcurrentlyStatus)ret;
            if (status == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
                AbstractSpan activeSpan = ContextManager.activeSpan();
                activeSpan.errorOccurred();
                Tags.STATUS_CODE.set(activeSpan, status.name());
            }
            ContextManager.stopSpan();
            return ret;
        }
    }
    复制代码

*  MessageConcurrentlyConsumeInterceptor继承了AbstractMessageConsumeInterceptor，其afterMethod方法获取ConsumeConcurrentlyStatus，在该值为ConsumeConcurrentlyStatus.RECONSUME\_LATER时执行activeSpan.errorOccurred()并设置STATUS\_CODE，最后执行ContextManager.stopSpan()

### ConsumeMessageOrderlyInstrumentation ###

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java

publicclassConsumeMessageOrderlyInstrumentationextendsClassInstanceMethodsEnhancePluginDefine {
        privatestatic final StringENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly";
        privatestatic final StringENHANCE_METHOD = "consumeMessage";
        privatestatic final StringINTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageOrderlyConsumeInterceptor";
    
        @OverridepublicConstructorInterceptPoint[] getConstructorsInterceptPoints() {
            returnnewConstructorInterceptPoint[0];
        }
    
        @OverridepublicInstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
            returnnewInstanceMethodsInterceptPoint[] {
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(ENHANCE_METHOD);
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnINTERCEPTOR_CLASS;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                }
            };
        }
    
        @OverrideprotectedClassMatchenhanceClass() {
            returnbyHierarchyMatch(newString[] {ENHANCE_CLASS});
        }
    }
    复制代码

*  ConsumeMessageOrderlyInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageOrderlyConsumeInterceptor增强org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly的consumeMessage方法

#### MessageOrderlyConsumeInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/MessageOrderlyConsumeInterceptor.java

public classMessageOrderlyConsumeInterceptorextendsAbstractMessageConsumeInterceptor{
    
        @Override
        public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret) throwsThrowable {
    
            ConsumeOrderlyStatus status = (ConsumeOrderlyStatus)ret;
            if (status == ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT) {
                AbstractSpan activeSpan = ContextManager.activeSpan();
                activeSpan.errorOccurred();
                Tags.STATUS_CODE.set(activeSpan, status.name());
            }
            ContextManager.stopSpan();
            return ret;
        }
    
    }
    复制代码

*  MessageOrderlyConsumeInterceptor继承了AbstractMessageConsumeInterceptor，其afterMethod方法获取ConsumeOrderlyStatus，在该值为ConsumeOrderlyStatus.SUSPEND\_CURRENT\_QUEUE\_A\_MOMENT时执行activeSpan.errorOccurred()并设置STATUS\_CODE，最后执行ContextManager.stopSpan()

### MQClientAPIImplInstrumentation ###

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java

publicclassMQClientAPIImplInstrumentationextendsClassInstanceMethodsEnhancePluginDefine {
    
        privatestatic final StringENHANCE_CLASS = "org.apache.rocketmq.client.impl.MQClientAPIImpl";
        privatestatic final StringSEND_MESSAGE_METHOD_NAME = "sendMessage";
        privatestatic final StringASYNC_METHOD_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageSendInterceptor";
        publicstatic final StringUPDATE_NAME_SERVER_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v4.UpdateNameServerInterceptor";
        publicstatic final StringUPDATE_NAME_SERVER_METHOD_NAME = "updateNameServerAddressList";
    
        @OverridepublicConstructorInterceptPoint[] getConstructorsInterceptPoints() {
            returnnewConstructorInterceptPoint[0];
        }
    
        @OverridepublicInstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
            returnnewInstanceMethodsInterceptPoint[] {
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(SEND_MESSAGE_METHOD_NAME).and(takesArguments(12));
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnASYNC_METHOD_INTERCEPTOR;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                },
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(UPDATE_NAME_SERVER_METHOD_NAME);
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnUPDATE_NAME_SERVER_INTERCEPT_CLASS;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                }
            };
        }
    
        @OverrideprotectedClassMatchenhanceClass() {
            returnbyName(ENHANCE_CLASS);
        }
    }
    复制代码

*  MQClientAPIImplInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它增强的是org.apache.rocketmq.client.impl.MQClientAPIImpl类；它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageSendInterceptor增强其sendMessage方法；它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.UpdateNameServerInterceptor增强其updateNameServerAddressList方法

#### MessageSendInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java

public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor {
    
        public static final String ASYNC_SEND_OPERATION_NAME_PREFIX = "RocketMQ/";
    
        @Override
        public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            MethodInterceptResult result) throws Throwable {
            Message message = (Message)allArguments[2];
            ContextCarrier contextCarrier = new ContextCarrier();
            String namingServiceAddress = String.valueOf(objInst.getSkyWalkingDynamicField());
            AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress);
            span.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
            Tags.MQ_BROKER.set(span, (String)allArguments[0]);
            Tags.MQ_TOPIC.set(span, message.getTopic());
            SpanLayer.asMQ(span);
    
            SendMessageRequestHeader requestHeader = (SendMessageRequestHeader)allArguments[3];
            StringBuilder properties = new StringBuilder(requestHeader.getProperties());
            CarrierItem next = contextCarrier.items();
            while (next.hasNext()) {
                next = next.next();
                if (!StringUtil.isEmpty(next.getHeadValue())) {
                    properties.append(next.getHeadKey());
                    properties.append(NAME_VALUE_SEPARATOR);
                    properties.append(next.getHeadValue());
                    properties.append(PROPERTY_SEPARATOR);
                }
            }
            requestHeader.setProperties(properties.toString());
    
            if (allArguments[6] != null) {
                ((EnhancedInstance)allArguments[6]).setSkyWalkingDynamicField(new SendCallBackEnhanceInfo(message.getTopic(), ContextManager.capture()));
            }
        }
    
        @Override
        public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret) throws Throwable {
            ContextManager.stopSpan();
            return ret;
        }
    
        @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
            Class<?>[] argumentsTypes, Throwable t) {
            ContextManager.activeSpan().errorOccurred().log(t);
        }
    
        private String buildOperationName(String topicName) {
            return ASYNC_SEND_OPERATION_NAME_PREFIX + topicName + "/Producer";
        }
    }
    复制代码

*  MessageSendInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法设置MQ\_BROKER、MQ\_TOPIC，并将contextCarrier.items()放到SendMessageRequestHeader的properties进行透传；其afterMethod方法执行ContextManager.stopSpan()；其handleMethodException方法执行ContextManager.activeSpan().errorOccurred().log(t)

#### UpdateNameServerInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java

publicclassUpdateNameServerInterceptorimplementsInstanceMethodsAroundInterceptor {
        @OverridepublicvoidbeforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            MethodInterceptResult result) throws Throwable {
            objInst.setSkyWalkingDynamicField(allArguments[0]);
        }
    
        @OverridepublicObjectafterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret) throws Throwable {
            return ret;
        }
    
        @OverridepublicvoidhandleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
            Class<?>[] argumentsTypes, Throwable t) {
    
        }
    }
    复制代码

*  UpdateNameServerInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法执行objInst.setSkyWalkingDynamicField(allArguments\[0\])

### SendCallbackInstrumentation ###

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/SendCallbackInstrumentation.java

publicclassSendCallbackInstrumentationextendsClassInstanceMethodsEnhancePluginDefine {
    
        privatestatic final StringENHANCE_CLASS = "org.apache.rocketmq.client.producer.SendCallback";
        privatestatic final StringON_SUCCESS_ENHANCE_METHOD = "onSuccess";
        privatestatic final StringON_SUCCESS_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v4.OnSuccessInterceptor";
        privatestatic final StringON_EXCEPTION_METHOD = "onException";
        privatestatic final StringON_EXCEPTION_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v4.OnExceptionInterceptor";
    
        @OverridepublicConstructorInterceptPoint[] getConstructorsInterceptPoints() {
            returnnewConstructorInterceptPoint[0];
        }
    
        @OverridepublicInstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
            returnnewInstanceMethodsInterceptPoint[] {
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(ON_SUCCESS_ENHANCE_METHOD).and(takesArgumentWithType(0, "org.apache.rocketmq.client.producer.SendResult"));
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnON_SUCCESS_INTERCEPTOR;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                },
                newInstanceMethodsInterceptPoint() {
                    @OverridepublicElementMatcher<MethodDescription> getMethodsMatcher() {
                        returnnamed(ON_EXCEPTION_METHOD).and(takesArgumentWithType(0, "java.lang.Throwable"));
                    }
    
                    @OverridepublicStringgetMethodsInterceptor() {
                        returnON_EXCEPTION_INTERCEPTOR;
                    }
    
                    @OverridepublicbooleanisOverrideArgs() {
                        returnfalse;
                    }
                }
            };
        }
    
        @OverrideprotectedClassMatchenhanceClass() {
            returnbyHierarchyMatch(newString[] {ENHANCE_CLASS});
        }
    }
    复制代码

*  SendCallbackInstrumentation继承了ClassInstanceMethodsEnhancePluginDefine，它增强的是org.apache.rocketmq.client.producer.SendCallback类；它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.OnSuccessInterceptor增强其第一个参数类型为org.apache.rocketmq.client.producer.SendResult的onSuccess方法；它使用org.apache.skywalking.apm.plugin.rocketMQ.v4.OnExceptionInterceptor增强其第一个参数类型为java.lang.Throwable的onException方法

#### OnSuccessInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnSuccessInterceptor.java

public class OnSuccessInterceptor implements InstanceMethodsAroundInterceptor {
    
        public static final String CALLBACK_OPERATION_NAME_PREFIX = "RocketMQ/";
    
        @Override
        public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            MethodInterceptResult result) throws Throwable {
            SendCallBackEnhanceInfo enhanceInfo = (SendCallBackEnhanceInfo)objInst.getSkyWalkingDynamicField();
            AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + enhanceInfo.getTopicId() + "/Producer/Callback");
            activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
            SendStatus sendStatus = ((SendResult)allArguments[0]).getSendStatus();
            if (sendStatus != SendStatus.SEND_OK) {
                activeSpan.errorOccurred();
                Tags.STATUS_CODE.set(activeSpan, sendStatus.name());
            }
            ContextManager.continued(enhanceInfo.getContextSnapshot());
        }
    
        @Override
        public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret) throws Throwable {
            ContextManager.stopSpan();
            return ret;
        }
    
        @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
            Class<?>[] argumentsTypes, Throwable t) {
            ContextManager.activeSpan().errorOccurred().log(t);
        }
    }
    复制代码

*  OnSuccessInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法执行ContextManager.continued(enhanceInfo.getContextSnapshot())；其afterMethod方法执行ContextManager.stopSpan()；其handleMethodException方法执行ContextManager.activeSpan().errorOccurred().log(t)

#### OnExceptionInterceptor ####

skywalking-6.6.0/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/OnExceptionInterceptor.java

publicclassOnExceptionInterceptorimplementsInstanceMethodsAroundInterceptor {
    
        publicstaticfinalStringCALLBACK_OPERATION_NAME_PREFIX="RocketMQ/";
    
        @OverridepublicvoidbeforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            MethodInterceptResult result)throws Throwable {
            SendCallBackEnhanceInfoenhanceInfo= (SendCallBackEnhanceInfo)objInst.getSkyWalkingDynamicField();
            AbstractSpanactiveSpan= ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + enhanceInfo.getTopicId() + "/Producer/Callback");
            activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
            activeSpan.errorOccurred().log((Throwable)allArguments[0]);
            ContextManager.continued(enhanceInfo.getContextSnapshot());
        }
    
        @Overridepublic Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
            Object ret)throws Throwable {
            ContextManager.stopSpan();
            return ret;
        }
    
        @OverridepublicvoidhandleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
            Class<?>[] argumentsTypes, Throwable t) {
            ContextManager.activeSpan().log(t);
        }
    }
    复制代码

*  OnExceptionInterceptor实现了InstanceMethodsAroundInterceptor接口，其beforeMethod方法执行activeSpan.errorOccurred().log((Throwable)allArguments\[0\])以及ContextManager.continued(enhanceInfo.getContextSnapshot())；其afterMethod方法执行ContextManager.stopSpan()；其handleMethodException方法执行ContextManager.activeSpan().log(t)方法

### 小结 ###

skywalking的rocketmq-plugin定义了4个增强，分别是ConsumeMessageConcurrentlyInstrumentation、ConsumeMessageOrderlyInstrumentation、MQClientAPIImplInstrumentation、SendCallbackInstrumentation

作者：go4it

原文链接：[https://juejin.cn/post/6844904088409210887][https_juejin.cn_post_6844904088409210887]

[https_juejin.cn_post_6844904088409210887]: https://juejin.cn/post/6844904088409210887