dubbo監控機制之監控數據上報分析

dubbo對服務運行的監控，是經過從provider和consumer方收集調用信息存盤後，再由監控中心對數據分析繪表的方式完成的。
具體實現是provider和consumer向監控中心推數據。
今天以服務消費方爲例，經過源碼分析下消費方向監控中心上報數據的過程。
配置監控中心的兩種方式：

<!--1,表示從註冊中心發現監控中心地址-->
<dubbo:monitor protocol="registry"></dubbo:monitor>
<!--2,直連監控中心服務器地址-->
<dubbo:monitor address="10.47.17.170"></dubbo:monitor>

<!--配置過濾器monitor，dubbo是經過過濾器實現調用信息上報的-->
<dubbo:reference id="demoService"  interface="demo.dubbo.api.DemoService"   timeout="6000" filter="monitor"/>

以上spring配置裏的<dubbo:monitor>標籤的解析，在ReferenceBean的afterPropertiesSet方法中，邏輯以下

public void afterPropertiesSet() throws Exception {
   
   //....其餘代碼略
   if (getMonitor() == null
                && (getConsumer() == null || getConsumer().getMonitor() == null)
                && (getApplication() == null || getApplication().getMonitor() == null)) {
		//解析MonitorConfig類，從容器中獲取monitorConfig對象
            Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
            if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
                MonitorConfig monitorConfig = null;
                for (MonitorConfig config : monitorConfigMap.values()) {
                    if (config.isDefault() == null || config.isDefault().booleanValue()) {
                        if (monitorConfig != null) {
                            throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
                        }
                        monitorConfig = config;
                    }
                }
		//把解析後對象賦值給monitor屬性，後面構造代理會用到
                if (monitorConfig != null) {
                    setMonitor(monitorConfig);
                }
            }
        }
   }

在構造代理邏輯在ReferenceConfig類的createProxy方法中，由於咱們這裏走註冊中心發現監控中心，因此看下面一段邏輯：

//構造註冊中心url
	List<URL> us = loadRegistries(false);
	if (us != null && us.size() > 0) {
	    for (URL u : us) {
		//經過註冊中心的url構造monitor Url(***跟蹤下loadMonitor***)
		URL monitorUrl = loadMonitor(u);
		if (monitorUrl != null) {
		    //放置監控url到map key爲「monitor」(***重點在這裏***)
		    map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
		}
		urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
	    }
	}
	if (urls == null || urls.size() == 0) {
	    throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
	}

跟到AbstractInterfaceConfig類的loadMonitor方法：

/***
     * 構造監控中心URL
     * @param registryURL
     * @return
     */
    protected URL loadMonitor(URL registryURL) {
        if (monitor == null) {
            //沒有配置監控中心，從dubbo.monitor.address屬性中獲取
            String monitorAddress = ConfigUtils.getProperty("dubbo.monitor.address");
            //獲取監控中心服務發現協議，好比經過註冊中心
            String monitorProtocol = ConfigUtils.getProperty("dubbo.monitor.protocol");
            if (monitorAddress != null && monitorAddress.length() > 0
                    || monitorProtocol != null && monitorProtocol.length() > 0) {
                //都沒有配置，new一個對象
                monitor = new MonitorConfig();
            } else {
                //沒有註冊中心
                return null;
            }
        }
        //把屬性文件中的的值，填充到monitor對象裏
        appendProperties(monitor);
        Map<String, String> map = new HashMap<String, String>();
        //
        //這裏接口固定是MonitorService.class.getName(),就是固定經過這個接口提供服務上報服務
	//這裏的MonitorService服務是由監控中心實現並註冊的到註冊中心。
        map.put(Constants.INTERFACE_KEY, MonitorService.class.getName());
        map.put("dubbo", Version.getVersion());
        map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
        if (ConfigUtils.getPid() > 0) {
            map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
        }
        //把monitor對象裏的屬性，放到map裏去，key是對象屬性名
        appendParameters(map, monitor);
        String address = monitor.getAddress();
        String sysaddress = System.getProperty("dubbo.monitor.address");
        if (sysaddress != null && sysaddress.length() > 0) {
            address = sysaddress;
        }
        //設置監控protocal
        if (ConfigUtils.isNotEmpty(address)) {
            if (!map.containsKey(Constants.PROTOCOL_KEY)) {
                if (ExtensionLoader.getExtensionLoader(MonitorFactory.class).hasExtension("logstat")) {
                    map.put(Constants.PROTOCOL_KEY, "logstat");
                } else {//沒有logstat spi擴展，就用dubbo協議
                    map.put(Constants.PROTOCOL_KEY, "dubbo");
                }
            }
            //構造經過address和map，構造url
            return UrlUtils.parseURL(address, map);
        } else if (Constants.REGISTRY_PROTOCOL.equals(monitor.getProtocol()) && registryURL != null) {
            //若是monitor配置是經過註冊中心發現，監控服務，設置protocol是dubbo, 添加參數 protocol=registry,refer=StringUtils.toQueryString(map)
            return registryURL.setProtocol("dubbo").addParameter(Constants.PROTOCOL_KEY, "registry").addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map));
        }
        return null;
    }

以上邏輯構造了monitorUrl並經過 monitor key放入url的參數中。
因爲dubbo是經過過濾器上報監控數據的，(關於dubbo使用過濾器機制，還要從dubbo aop實現入手),下面分析下具體過濾器如何使用monitorUrl的，能夠看懂文章開始咱們配置的過濾器是「monitor」
因此這裏，看下Filter的monitor spi實現,MonitorFilter類，具體在invoke方法裏:

//調用過程攔截
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        if (invoker.getUrl().hasParameter(Constants.MONITOR_KEY)) {
            RpcContext context = RpcContext.getContext(); // 提供方必須在invoke()以前獲取context信息
            String remoteHost = context.getRemoteHost();
            long start = System.currentTimeMillis(); // 記錄起始時間戮
            getConcurrent(invoker, invocation).incrementAndGet(); // 併發計數++
            try {
                Result result = invoker.invoke(invocation); // 讓調用鏈往下執行
                //上報調用統計信息（***看這裏**）
                collect(invoker, invocation, result, remoteHost, start, false);
                return result;
            } catch (RpcException e) {
                collect(invoker, invocation, null, remoteHost, start, true);
                throw e;
            } finally {
                getConcurrent(invoker, invocation).decrementAndGet(); // 併發計數++
            }
        } else {
            return invoker.invoke(invocation);
        }
    }

//具體
private void collect(Invoker<?> invoker, Invocation invocation, Result result, String remoteHost, long start, boolean error) {
        try {
            // ---- 服務信息獲取 ----
            long elapsed = System.currentTimeMillis() - start; // 計算調用耗時
            int concurrent = getConcurrent(invoker, invocation).get(); // 當前併發數
            String application = invoker.getUrl().getParameter(Constants.APPLICATION_KEY);
            String service = invoker.getInterface().getName(); // 獲取服務名稱
            String method = RpcUtils.getMethodName(invocation); // 獲取方法名
           
            URL url = invoker.getUrl().getUrlParameter(Constants.MONITOR_KEY);
	     //經過 monitor key 獲取監控url （***看這裏**），這裏monitorFactory是spi機制生成的MonitorFactory$Adaptive
	     //這裏實際是走的DubboMonitorFactroy類的getMonitor方法
            Monitor monitor = monitorFactory.getMonitor(url);
            int localPort;
            String remoteKey;
            String remoteValue;
            if (Constants.CONSUMER_SIDE.equals(invoker.getUrl().getParameter(Constants.SIDE_KEY))) {
                // ---- 服務消費方監控 ----
                localPort = 0;
                remoteKey = MonitorService.PROVIDER;
                remoteValue = invoker.getUrl().getAddress();
            } else {
                // ---- 服務提供方監控 ----
                localPort = invoker.getUrl().getPort();
                remoteKey = MonitorService.CONSUMER;
                remoteValue = remoteHost;
            }
            String input = "", output = "";
            if (invocation.getAttachment(Constants.INPUT_KEY) != null) {
                input = invocation.getAttachment(Constants.INPUT_KEY);
            }
            if (result != null && result.getAttachment(Constants.OUTPUT_KEY) != null) {
                output = result.getAttachment(Constants.OUTPUT_KEY);
            }
            //經過上面構造的監控上報工具，上報數據（***看這裏**）
            monitor.collect(new URL(Constants.COUNT_PROTOCOL,
                    NetUtils.getLocalHost(), localPort,
                    service + "/" + method,
                    MonitorService.APPLICATION, application,
                    MonitorService.INTERFACE, service,
                    MonitorService.METHOD, method,
                    remoteKey, remoteValue,
                    error ? MonitorService.FAILURE : MonitorService.SUCCESS, "1",
                    MonitorService.ELAPSED, String.valueOf(elapsed),
                    MonitorService.CONCURRENT, String.valueOf(concurrent),
                    Constants.INPUT_KEY, input,
                    Constants.OUTPUT_KEY, output));
        } catch (Throwable t) {
            logger.error("Failed to monitor count service " + invoker.getUrl() + ", cause: " + t.getMessage(), t);
        }
    }

看下DubboMonitorFactroy類的getMonitor方法，實如今其父類AbstractMonitorFactory中:

public Monitor getMonitor(URL url) {
        //這裏設置上報服務接口MonitorService
        url = url.setPath(MonitorService.class.getName()).addParameter(Constants.INTERFACE_KEY, MonitorService.class.getName());
        String key = url.toServiceStringWithoutResolving();
        LOCK.lock();
        try {
            //從緩存中獲取
            Monitor monitor = MONITORS.get(key);
            if (monitor != null) {
                return monitor;
            }
            //經過url建立monitor,在子類DubboMonitorFactroy中實現
            monitor = createMonitor(url);
            if (monitor == null) {
                throw new IllegalStateException("Can not create monitor " + url);
            }
            MONITORS.put(key, monitor);
            return monitor;
        } finally {
            // 釋放鎖
            LOCK.unlock();
        }
    }

DubboMonitorFactroy裏實現的createMonitor方法:

protected Monitor createMonitor(URL url) {
        //這裏會經過url的protocol參數獲取協議值，若是是經過註冊中心發現監控中心服務的方式，這裏
        //protocol的值是registry，不然就是dubbo
        url = url.setProtocol(url.getParameter(Constants.PROTOCOL_KEY, "dubbo"));
        if (url.getPath() == null || url.getPath().length() == 0) {
            url = url.setPath(MonitorService.class.getName());
        }
        String filter = url.getParameter(Constants.REFERENCE_FILTER_KEY);
        if (filter == null || filter.length() == 0) {
            filter = "";
        } else {
            filter = filter + ",";
        }
        //監控中心服務配置多個的場景，這裏默認使用failsafe容錯機制
        url = url.addParameters(Constants.CLUSTER_KEY, "failsafe", Constants.CHECK_KEY, String.valueOf(false),
                Constants.REFERENCE_FILTER_KEY, filter + "-monitor");
        //這裏protocol也是Protocol$Adpative的，若是協議是registry 要走經過註冊中心發現服務那一套邏輯。
        Invoker<MonitorService> monitorInvoker = protocol.refer(MonitorService.class, url);
        //建立服務代理代理
        MonitorService monitorService = proxyFactory.getProxy(monitorInvoker);
        //最後構造BubboMonitor對象
        return new DubboMonitor(monitorInvoker, monitorService);
    }

這裏看下DubboMonitor類繼承圖，能夠看到它實現了MonitorService接口

//構造函數
public DubboMonitor(Invoker<MonitorService> monitorInvoker, MonitorService monitorService) {
        this.monitorInvoker = monitorInvoker;
        this.monitorService = monitorService;
        this.monitorInterval = monitorInvoker.getUrl().getPositiveParameter("interval", 60000);
        // 啓動統計信息收集定時器，設置上報頻率monitorInterval，因此說，上報數據是異步的
        sendFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                // send方法收集統計信息
                try {
		    (***看這裏***)
                    send();
                } catch (Throwable t) { // 防護性容錯
                    logger.error("Unexpected error occur at send statistic, cause: " + t.getMessage(), t);
                }
            }
        }, monitorInterval, monitorInterval, TimeUnit.MILLISECONDS);
    }

     //從本地靜態變量中獲取統計信息，經過遠程服務monitorService接口方法上報。
    public void send() {
        if (logger.isInfoEnabled()) {
            logger.info("Send statistics to monitor " + getUrl());
        }
        String timestamp = String.valueOf(System.currentTimeMillis());
        for (Map.Entry<Statistics, AtomicReference<long[]>> entry : statisticsMap.entrySet()) {
            // 獲取已統計數據
            Statistics statistics = entry.getKey();
            AtomicReference<long[]> reference = entry.getValue();
            long[] numbers = reference.get();
            long success = numbers[0];
            long failure = numbers[1];
            long input = numbers[2];
            long output = numbers[3];
            long elapsed = numbers[4];
            long concurrent = numbers[5];
            long maxInput = numbers[6];
            long maxOutput = numbers[7];
            long maxElapsed = numbers[8];
            long maxConcurrent = numbers[9];

            // 發送彙總信息
            URL url = statistics.getUrl()
                    .addParameters(MonitorService.TIMESTAMP, timestamp,
                            MonitorService.SUCCESS, String.valueOf(success),
                            MonitorService.FAILURE, String.valueOf(failure),
                            MonitorService.INPUT, String.valueOf(input),
                            MonitorService.OUTPUT, String.valueOf(output),
                            MonitorService.ELAPSED, String.valueOf(elapsed),
                            MonitorService.CONCURRENT, String.valueOf(concurrent),
                            MonitorService.MAX_INPUT, String.valueOf(maxInput),
                            MonitorService.MAX_OUTPUT, String.valueOf(maxOutput),
                            MonitorService.MAX_ELAPSED, String.valueOf(maxElapsed),
                            MonitorService.MAX_CONCURRENT, String.valueOf(maxConcurrent)
                    );
            //調用監控中心發佈的MonitorService服務，上報調用統計信息
            monitorService.collect(url);

            // 減掉已統計數據
            long[] current;
            long[] update = new long[LENGTH];
            do {
                current = reference.get();
                if (current == null) {
                    update[0] = 0;
                    update[1] = 0;
                    update[2] = 0;
                    update[3] = 0;
                    update[4] = 0;
                    update[5] = 0;
                } else {
                    update[0] = current[0] - success;
                    update[1] = current[1] - failure;
                    update[2] = current[2] - input;
                    update[3] = current[3] - output;
                    update[4] = current[4] - elapsed;
                    update[5] = current[5] - concurrent;
                }
            } while (!reference.compareAndSet(current, update));
        }
    }

    //而DubboMonitor自己的collect方法，供信息上報處，過濾器中調用
    //每次的調用信息，放入本地靜態變量statisticsMap中，
    public void collect(URL url) {
        // 讀寫統計變量
        int success = url.getParameter(MonitorService.SUCCESS, 0);
        int failure = url.getParameter(MonitorService.FAILURE, 0);
        int input = url.getParameter(MonitorService.INPUT, 0);
        int output = url.getParameter(MonitorService.OUTPUT, 0);
        int elapsed = url.getParameter(MonitorService.ELAPSED, 0);
        int concurrent = url.getParameter(MonitorService.CONCURRENT, 0);
        // 初始化原子引用
        Statistics statistics = new Statistics(url);
        AtomicReference<long[]> reference = statisticsMap.get(statistics);
        if (reference == null) {
            statisticsMap.putIfAbsent(statistics, new AtomicReference<long[]>());
            reference = statisticsMap.get(statistics);
        }
        // CompareAndSet併發加入統計數據
        long[] current;
        long[] update = new long[LENGTH];
        do {
            current = reference.get();
            if (current == null) {
                update[0] = success;
                update[1] = failure;
                update[2] = input;
                update[3] = output;
                update[4] = elapsed;
                update[5] = concurrent;
                update[6] = input;
                update[7] = output;
                update[8] = elapsed;
                update[9] = concurrent;
            } else {
                update[0] = current[0] + success;
                update[1] = current[1] + failure;
                update[2] = current[2] + input;
                update[3] = current[3] + output;
                update[4] = current[4] + elapsed;
                update[5] = (current[5] + concurrent) / 2;
                update[6] = current[6] > input ? current[6] : input;
                update[7] = current[7] > output ? current[7] : output;
                update[8] = current[8] > elapsed ? current[8] : elapsed;
                update[9] = current[9] > concurrent ? current[9] : concurrent;
            }
        } while (!reference.compareAndSet(current, update));
    }

以上梳理了下，服務消費方配置監控中心並上報調用數據的流程，
服務提供方上報監控中心的流程是同樣的。一樣使用這個過濾器完成。
下次再梳理下，監控中心自己的處理邏輯。