快捷導(dǎo)航

SkyWalking?自定義插件(Spring?RabbitMQ)具體分析過程

更新時(shí)間：2022年02月14日 11:52:36 作者：Gonk

這篇文章主要介紹了SkyWalking?自定義插件(Spring?RabbitMQ)具體分析過程，本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下

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

RabbitMQ插件問題

skywalking官方提供的RabbitMQ插件存在缺陷，其只針對(duì)RabbitMQ官方原生Client實(shí)現(xiàn)擴(kuò)展，但我們?cè)陧?xiàng)目中一般不直接使用原生Client，而是使用Spring RabitMQ Client，因Spring RabitMQ Consumer中存在跨線程操作，導(dǎo)致跟蹤ID斷鏈。

具體分析過程

1.官方插件源碼的攔截點(diǎn)是原生Consumer的handleDelivery方法，源碼如下:

2.而Spring RabbitMQ消費(fèi)者的默認(rèn)實(shí)現(xiàn)是BlockingQueueConsumer, handleDelivery核心邏輯是把消息放到內(nèi)部的BlockingQueue隊(duì)列，不做真正的消費(fèi)處理，因此攔截此處無法關(guān)聯(lián)到消費(fèi)者邏輯，源碼如下

@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.真正的消費(fèi)處理在SimpleMessageListenerContainer，SimpleMessageListenerContainer繼承Runnable接口，在其run方法中while循環(huán)調(diào)用mainLoop方法，整體調(diào)用鏈路為

4.SimpleMessageListenerContainer.run() -> SimpleMessageListenerContainer.mainLoop() -> SimpleMessageListenerContainer.receiveAndExecute() -> SimpleMessageListenerContainer.doReceiveAndExecute() -> AbstractMessageListenerContainer.executeListener()。最終在executeListener中執(zhí)行消費(fèi)邏輯

protected void executeListener(Channel channel, Object data) {
	...
		try {
      // 執(zhí)行消費(fèi)邏輯
			doExecuteListener(channel, data);
			if (sample != null) {
				this.micrometerHolder.success(sample, data instanceof Message
						? ((Message) data).getMessageProperties().getConsumerQueue()
						: queuesAsListString());
			}
		}
		catch (RuntimeException ex) {
		....
		}
	}

實(shí)現(xiàn)自定義插件

從上面可以分析出，AbstractMessageListenerContainer.executeListener()是最佳的攔截點(diǎn)
實(shí)現(xiàn)源碼已放到碼云倉庫:https://gitee.com/eureka-gitee/apm-sniffer-pro/tree/v7.0.0.0/