skywalking官方提供的RabbitMQ插件存在缺陷,其只针对RabbitMQ官方原生Client实现扩展,但我们在项目中一般不直接使用原生Client,而是使用Spring RabitMQ Client,因Spring RabitMQ Consumer中存在跨线程操作,导致跟踪ID断链。
具体分析过程 官方插件源码的拦截点是原生Consumer的handleDelivery方法,源码如下:而Spring RabbitMQ消费者的默认实现是BlockingQueueConsumer, handleDelivery核心逻辑是把消息放到内部的BlockingQueue队列,不做真正的消费处理,因此拦截此处无法关联到消费者逻辑,源码如下
@Overridepublic 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 happenBlockingQueueConsumer.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);}}
真正的消费处理在SimpleMessageListenerContainer,SimpleMessageListenerContainer继承Runnable接口,在其run方法中while循环调用mainLoop方法,整体调用链路为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) {....}}
实现自定义插件 从上面可以分析出,AbstractMessageListenerContainer.executeListener()是最佳的拦截点
实现源码已放到码云仓库:https://gitee.com/eureka-gitee/apm-sniffer-pro/tree/v7.0.0.0/
SkyWalking调用链路
logback日志