导航：首页 > 开发技术 >

SkyWalking自定义插件怎么用

发表于：2025-01-22 作者：千家信息网编辑

千家信息网最后更新 2025年01月22日，这篇文章将为大家详细讲解有关SkyWalking自定义插件怎么用，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。RabbitMQ插件问题skywalking官方提供的

千家信息网最后更新 2025年01月22日SkyWalking自定义插件怎么用

这篇文章将为大家详细讲解有关SkyWalking自定义插件怎么用，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。

RabbitMQ插件问题

skywalking官方提供的RabbitMQ插件存在缺陷，其只针对RabbitMQ官方原生Client实现扩展，但我们在项目中一般不直接使用原生Client，而是使用Spring RabitMQ Client，因Spring RabitMQ Consumer中存在跨线程操作，导致跟踪ID断链。

具体分析过程

1.官方插件源码的拦截点是原生Consumer的handleDelivery方法，源码如下:

2.而Spring RabbitMQ消费者的默认实现是BlockingQueueConsumer, handleDelivery核心逻辑是把消息放到内部的BlockingQueue队列，不做真正的消费处理，因此拦截此处无法关联到消费者逻辑，源码如下

@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.真正的消费处理在SimpleMessageListenerContainer，SimpleMessageListenerContainer继承Runnable接口，在其run方法中while循环调用mainLoop方法，整体调用链路为

4.SimpleMessageListenerContainer.run() -> SimpleMessageListenerContainer.mainLoop() -> SimpleMessageListenerContainer.receiveAndExecute() -> SimpleMessageListenerContainer.doReceiveAndExecute() -> AbstractMessageListenerContainer.executeListener()。最终在executeListener中执行消费逻辑

protected void executeListener(Channel channel, Object data) {        ...                try {      // 执行消费逻辑                        doExecuteListener(channel, data);                        if (sample != null) {                                this.micrometerHolder.success(sample, data instanceof Message                                                ? ((Message) data).getMessageProperties().getConsumerQueue()                                                : queuesAsListString());                        }                }                catch (RuntimeException ex) {                ....                }        }