快捷導(dǎo)航

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

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

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

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

RabbitMQ插件問題

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

具體分析過程

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

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

@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循環(huán)調(diào)用mainLoop方法，整體調(diào)用鏈路為

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

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