SkyWalking 自定义插件(Spring RabbitMQ)具体分析过程

SkyWalking 自定义插件(Spring RabbitMQ)具体分析过程

SkyWalking 自定义插件(Spring RabbitMQ) 官方

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(

http:// 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

}

}http://

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.doReceivMhuUEScyeAndExecute() -> 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) {

....

}

}

实现自定义插件

从上面可以分析出，AbstractMessageListenerContainer.executeListener()是最佳的拦截点实现源码已放到码云仓库:https://gitee.com/eureka-gitee/apm-sniffer-pro/tree/v7.0.0.0/

效果展示

SkyWalking调用链路

logback日志

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。