深入理解SkyAPM

一、簡介

SkyAPM是提供分布式應用程序性能管理監控解決方案的開源項目。它可以跟蹤整個分布式系統的調用鏈，提供分布式事務追蹤，應用程序性能管理和分析，以支持運維人員進行問題排查和性能優化。

二、使用

SkyAPM的使用需要引入SkyWalking的Agent和Collector，分別用於在本地應用收集性能指標數據，和將收集到的數據上報到服務端進行分析處理與可視化。

  
    # 引入SkyWalking的Agent
    java -javaagent:/path/to/skywalking-agent.jar -Dskywalking.agent.service_name=myapp -jar myapp.jar
  

  
    # 啟動SkyWalking OAP Server作為收集器以處理數據並生成報表
    docker run --name skywalking -d -p 12800:12800 apache/skywalking-oap-server:8.5.0-es7
  

三、核心概念

1. 服務實例

服務實例是指運行着同一服務的單個應用程序或進程。SkyAPM跟蹤服務實例之間的請求和響應，以分析整個分布式系統的性能。

  
    // Java Spring Boot應用的服務實例配置
    spring.application.name=myapp
    server.port=8080
  

2. Spans

Span是指一次請求的一部分處理或事件，例如HTTP請求處理、SQL查詢、RPC調用等。每個Span都包含一個時間戳和其他屬性，可以用於調試、故障排除和性能優化。

  
    Span span = tracer.createLocalSpan("http: /api/user/1");
    span.tag("url", "/api/user/1");
    span.start();
    ...
    span.finish();
  

3. Traces

Trace是多個Span的有向無環圖，代表整個請求的調用鏈，包括客戶端和服務器端的Span。SkyAPM利用Trace來給一個完整的請求進行性能分析，以便檢測並診斷問題。

  
    val traceSegment: TraceSegmentObject = ContextManager.capture()
    segmentObserver.beforeSerialize(traceSegment)
    buffer.write(traceSegment.toByteArray())
  

四、插件開發

SkyAPM提供了各種插件，例如Kafka、Redis、Dubbo等，以便於您更好地監控和分析分布式系統的各個方面。插件API可以讓您擴展SkyAPM並添加自己的插件，以支持監控不同類型的應用程序和服務。

以下代碼是一個自定義的SkyWalking插件樣例，用於跟蹤ActiveMQ的生產者和消費者的消息發送和接收：

  
    @Component
    public class AmqInstrumentation implements SpanObserver, EnhanceRequireObjectCache {
      private static final Logger logger = LoggerFactory.getLogger(AmqInstrumentation.class);
    
      private static final String SEND_OPERATION_NAME = "activemq-send";
      private static final String CONSUME_OPERATION_NAME = "activemq-consume";
    
      private final ConcurrentMap sendCache = new ConcurrentHashMap();
      private final ConcurrentMap consumeCache = new ConcurrentHashMap();
    
      @Override
      public void afterFinished(Span span) {
        if (span.getOperationName().equals(SEND_OPERATION_NAME)) {
          SendCache cache = sendCache.remove(span.getSpanId());
          if (cache != null) {
            span.tag("activemq-broker-url", cache.brokerUrl);
            span.tag("activemq-destination", cache.destination);
            span.tag("activemq-message-queue-size", String.valueOf(cache.size));
          }
        } else if (span.getOperationName().equals(CONSUME_OPERATION_NAME)) {
          ConsumeCache cache = consumeCache.remove(span.getSpanId());
          if (cache != null) {
            span.tag("activemq-broker-url", cache.brokerUrl);
            span.tag("activemq-destination", cache.destination);
          }
        }
      }
    
      public void onSend(@Advice.Local("sendCache") SendCache cache,
                         @FieldName(className = "org.apache.activemq.command.ActiveMQDestination", value = "destinationType") byte destinationType,
                         @FieldValue(className = "org.apache.activemq.command.ActiveMQDestination", value = "physicalName") String physicalName,
                         @TargetObject Object messageProducer) {
        Span span = ContextManager.createExitSpan(SEND_OPERATION_NAME, ContextManager.getRemotePeer(messageProducer));
        if (span == null) {
          return;
        }
    
        cache.size = ((MessageProducer) messageProducer).getSendQueue().size();
        cache.brokerUrl = JmsHelper.getBrokerUrlFromConnection(JmsHelper.getJmsConnectionFromSession(JmsHelper.getSessionFromProducer((MessageProducer) messageProducer)));
        cache.destination = physicalName;
        cache.spanId = span.getSpanId();
        sendCache.put(span.getSpanId(), cache);
        ContextManager.continued(span);
      }
    
      public void onConsume(@Advice.Local("consumeCache") ConsumeCache cache,
                            @FieldName(className = "org.apache.activemq.command.ActiveMQDestination", value = "destinationType") byte destinationType,
                            @FieldValue(className = "org.apache.activemq.command.ActiveMQDestination", value = "physicalName") String physicalName,
                            @TargetObject Object messageConsumer) {
        Span span = ContextManager.createEntrySpan(CONSUME_OPERATION_NAME, new ActiveMqConsumerPeer((MessageConsumer) messageConsumer));
        if (span == null) {
          return;
        }
        cache.brokerUrl = JmsHelper.getBrokerUrlFromConnection(JmsHelper.getJmsConnectionFromSession(JmsHelper.getSessionFromConsumer((MessageConsumer) messageConsumer)));
        cache.destination = physicalName;
        cache.spanId = span.getSpanId();
        consumeCache.put(span.getSpanId(), cache);
        ContextManager.continued(span);
      }
    
      static class SendCache {
        String brokerUrl;
        String destination;
        int size;
        int spanId;
      }
    
      static class ConsumeCache {
        String brokerUrl;
        String destination;
        int spanId;
      }
    }