Shulou(Shulou.com)06/01 Report--

This article will explain in detail how to use the SkyWalking custom plug-in. The editor thinks it is very practical, so I share it with you as a reference. I hope you can get something after reading this article.

RabbitMQ plug-in issu

The official RabbitMQ plug-in provided by skywalking has defects. It is only extended for the official native Client of RabbitMQ. However, we generally do not directly use native Client in our projects, but use Spring RabitMQ Client instead. Because of the cross-thread operation in Spring RabitMQ Consumer, the tracking ID is broken.

Specific analysis process

1. The interception point of the official plug-in source code is the native Consumer handleDelivery method. The source code is as follows:

two。 The default implementation of Spring RabbitMQ consumers is BlockingQueueConsumer, and the core logic of handleDelivery is to put messages on the internal BlockingQueue queue without real consumption processing, so the interceptor cannot be associated with consumer logic here. The source code is as follows

Override public void handleDelivery (String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte [] body) {... Try {if (BlockingQueueConsumer.this.abortStarted > 0) {if (! BlockingQueueConsumer.this.queue.offer (new Delivery (consumerTag, envelope, properties, body, this.queueName) BlockingQueueConsumer.this.shutdownTimeout, TimeUnit.MILLISECONDS) {Channel channelToClose = super.getChannel () RabbitUtils.setPhysicalCloseRequired (channelToClose, true); / / Defensive-should never happen BlockingQueueConsumer.this.queue.clear () If (! this.canceled) {RabbitUtils.cancel (channelToClose, consumerTag) } try {channelToClose.close () Catch (@ SuppressWarnings ("unused") TimeoutException e) {/ / no-op} Else {BlockingQueueConsumer.this.queue .put (new Delivery (consumerTag) Envelope, properties, body, this.queueName)) } catch (@ SuppressWarnings ("unused") InterruptedException e) {Thread.currentThread () .interrupt (); catch (Exception e) {BlockingQueueConsumer.logger.warn ("Unexpected exception during delivery", e);}

3. The real consumption processing is that SimpleMessageListenerContainer,SimpleMessageListenerContainer inherits the Runnable interface, and in its run method, while calls the mainLoop method in a loop, and the overall calling link is

4.SimpleMessageListenerContainer.run ()-> SimpleMessageListenerContainer.mainLoop ()-> SimpleMessageListenerContainer.receiveAndExecute ()-> SimpleMessageListenerContainer.doReceiveAndExecute ()-> AbstractMessageListenerContainer.executeListener (). Finally, the consumption logic is executed in executeListener

Protected void executeListener (Channel channel, Object data) {... Try {/ / execute consumption logic doExecuteListener (channel, data); if (sample! = null) {this.micrometerHolder.success (sample, data instanceof Message? ((Message) data). GetMessageProperties (). GetConsumerQueue (): queuesAsListString ();}} catch (RuntimeException ex) {. }} implement a custom plug-in

From the above, it can be concluded that AbstractMessageListenerContainer.executeListener () is the best interception point.

The source code for the implementation has been placed in Code Cloud Warehouse: https://gitee.com/eureka-gitee/apm-sniffer-pro/tree/v7.0.0.0/

Effect display

SkyWalking call Link

Logback log

This is the end of this article on "how to use the SkyWalking custom plug-in". I hope the above content can be of some help to you, so that you can learn more knowledge. if you think the article is good, please share it for more people to see.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.