springboot+RabbitMQ+InfluxDB+Grafara監(jiān)控實踐

更新時間：2018年07月24日 09:25:48 作者：zygfengyuwuzu

這篇文章主要介紹了springboot+RabbitMQ+InfluxDB+Grafara監(jiān)控實踐,小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

本文需要有相關(guān)spring boot 或spring cloud 相關(guān)微服務(wù)框架的基礎(chǔ)，如果您具備相關(guān)基礎(chǔ)可以很容易的實現(xiàn)下述過程!!!!!!! 希望

本文的所說對需要的您有所幫助

從這里我們開始進入閑聊階段。

大家都知道 spring boot整合了很多很多的第三方框架，我們這里就簡單討論和使用性能監(jiān)控和JVM監(jiān)控相關(guān)的東西。其他的本文不討論雖然有些關(guān)聯(lián)，所以開篇有說需要有相關(guān)spring boot框架基礎(chǔ)說了這么多廢話，下面真正進入主題。

這里首先給大家看下整體的數(shù)據(jù)流程圖，其中兩條主線一條是接口或方法性能監(jiān)控數(shù)據(jù)收集，還有一條是spring boot 微服務(wù)JVM相關(guān)指標(biāo)數(shù)據(jù)采集，最后都匯總到InfluxDB時序數(shù)據(jù)庫中在用數(shù)據(jù)展示工具Grafara進行數(shù)據(jù)展示或報警。

〇、基礎(chǔ)服務(wù)

基礎(chǔ)服務(wù)比較多，其中包括RabbitMQ，Eureka注冊中心，influxDB，Grafara（不知道這些東西請百度或谷歌一下了解相關(guān)知識），下面簡單說下各基礎(chǔ)服務(wù)的功能：

RabbitMQ 一款很流行的消息中間件，主要用它來收集spring boot應(yīng)用監(jiān)控性能相關(guān)信息，為什么是RabbitMQ而不是什么別的 kafka等等，因為測試方便性能也夠用，spring boot整合的夠完善。

Eureka 注冊中心，一般看過或用過spring cloud相關(guān)框架的都知道spring cloud注冊中心主要推薦使用Eureka！至于為什么不做過多討論不是本文主要討論的關(guān)注點。本文主要用來同步和獲取注冊到注冊中心的應(yīng)用的相關(guān)信息。

InfluxDB和Grafara為什么選這兩個，其他方案如 ElasticSearch 、Logstash 、Kibana，ELK的組合等！原因很顯然 influxDB是時序數(shù)據(jù)庫數(shù)據(jù)的壓縮比率比其他（ElasticSearch ）好的很多（當(dāng)然本人沒有實際測試過都是看一些文檔）。同時InfluxDB使用SQL非常類似mysql等關(guān)系型數(shù)據(jù)庫入門方便，Grafara工具可預(yù)警。等等?。。。。。。。。。?！

好了工具就簡單介紹到這里，至于這些工具怎么部署搭建請搭建先自行找資料學(xué)習(xí)，還是因為不是本文重點介紹的內(nèi)容，不深入討論。如果有docker相關(guān)基礎(chǔ)的童鞋可以直接下載個鏡像啟動起來做測試使用（本人就是使用docker啟動的上面的基礎(chǔ)應(yīng)用（Eureka除外））

一、被監(jiān)控的應(yīng)用

這里不多說被監(jiān)控應(yīng)用肯定是spring boot項目但是要引用一下相關(guān)包和相關(guān)注解以及修改相關(guān)配置文件

包引用，這些包是必須引用的

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
    </dependency>
     <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-sleuth-zipkin-stream</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-starter-hystrix</artifactId>
    </dependency>

簡單說下呢相關(guān)包的功能spring-cloud-starter-netflix-eureka-client用于注冊中心使用的包，spring-cloud-starter-stream-rabbit 發(fā)送RabbitMQ相關(guān)包，spring-boot-starter-actuator發(fā)布監(jiān)控相關(guān)rest接口包，

spring-cloud-starter-hystrix熔斷性能監(jiān)控相關(guān)包。

相關(guān)注解

@EnableHystrix//開啟性能監(jiān)控
@RefreshScope//刷新配置文件 與本章無關(guān)
@EnableAutoConfiguration
@EnableFeignClients//RPC調(diào)用與本章無關(guān)
@RestController
@SpringBootApplication
public class ServerTestApplication {
  protected final static Logger logger = LoggerFactory.getLogger(ServerTestApplication.class);

  public static void main(String[] args) {
    SpringApplication.run(ServerTestApplication.class, args);
  }
}

配置文件相關(guān)

hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds: 60000
hystrix.threadpool.default.coreSize: 100
spring:
 application:
  name: spring-cloud-server2-test
 rabbitmq:
  host: 10.10.12.21
  port: 5672
  username: user
  password: password

encrypt:
 failOnError: false
server:
 port: 8081
eureka:
 instance:
  appname: spring-cloud-server2-test
  prefer-ip-address: true
 client: 
  serviceUrl:
   defaultZone: http://IP:PORT/eureka/#注冊中心地址
  eureka-server-total-connections-per-host: 500
endpoints:
 refresh:
  sensitive: false
 metrics:
  sensitive: false
 dump:
  sensitive: false
 auditevents:
  sensitive: false
 features:
  sensitive: false
 mappings:
  sensitive: false
 trace:
  sensitive: false
 autoconfig:
  sensitive: false
 loggers:
  sensitive: false

簡單解釋一下endpoints下面相關(guān)配置，主要就是原來這些路徑是需要授權(quán)訪問的，通過配置讓這些路徑接口不再是敏感的需要授權(quán)訪問的接口這應(yīng)我們就可以輕松的訪問注冊到注冊中心的每個服務(wù)的響應(yīng)的接口。這里插一句接口性能需要在方法上面加上如下類似相關(guān)注解，然后才會有相關(guān)性能數(shù)據(jù)輸出

@Value("${name}")
  private String name;
  
  @HystrixCommand(commandProperties = {
      @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "20000") }, threadPoolProperties = {
          @HystrixProperty(name = "coreSize", value = "64") }, threadPoolKey = "test1")
  @GetMapping("/testpro1")
  public String getStringtest1(){
    
    return name;
  }

好了到這里你的應(yīng)用基本上就具備相關(guān)性能輸出的能力了。你可以訪問

如果是上圖的接口你的應(yīng)用基本OK，為什么是基本因為你截圖沒有體現(xiàn)性能信息發(fā)送RabbitMQ的相關(guān)信息。這個需要看日志，加入你失敗了評論區(qū)在討論。我們先關(guān)注主線。

好的spring boot 應(yīng)用就先說道這里。開始下一主題

二、性能指標(biāo)數(shù)據(jù)采集

剛才訪問http://IP:port/hystrix.stream這個顯示出來的信息就是借口或方法性能相關(guān)信息的輸出，如果上面都沒有問題的話數(shù)據(jù)應(yīng)該發(fā)送到了RabbitMQ上面了我們直接去RabbitMQ上面接收相關(guān)數(shù)據(jù)就可以了。

性能指標(biāo)數(shù)據(jù)的采集服務(wù)主要應(yīng)用以下包

<dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <!-- https://mvnrepository.com/artifact/com.github.miwurster/spring-data-influxdb -->
    <dependency>
      <groupId>org.influxdb</groupId>
      <artifactId>influxdb-java</artifactId>
      <version>2.8</version>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-autoconfigure</artifactId>
    </dependency>

直接貼代碼

package application;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * 
 * @author zyg
 *
 */
@SpringBootApplication
public class RabbitMQApplication {

  public static void main(String[] args) {
    SpringApplication.run(RabbitMQApplication.class, args);
  }

}

package application;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;  

/**
 * 
 * @author zyg
 *
 */
@Configuration
public class RabbitMQConfig {
  public final static String QUEUE_NAME = "spring-boot-queue";
  public final static String EXCHANGE_NAME = "springCloudHystrixStream";
  public final static String ROUTING_KEY = "#";

  // 創(chuàng)建隊列
  @Bean
  public Queue queue() {
    return new Queue(QUEUE_NAME);
  }

  // 創(chuàng)建一個 topic 類型的交換器
  @Bean
  public TopicExchange exchange() {
    return new TopicExchange(EXCHANGE_NAME);
  }

  // 使用路由鍵（routingKey）把隊列（Queue）綁定到交換器（Exchange）
  @Bean
  public Binding binding(Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
  }

  @Bean
  public ConnectionFactory connectionFactory() {
    //rabbitmq IP 端口號
    CachingConnectionFactory connectionFactory = new CachingConnectionFactory("IP", 5672);
    connectionFactory.setUsername("user");
    connectionFactory.setPassword("password");
    return connectionFactory;
  }

  @Bean
  public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    return new RabbitTemplate(connectionFactory);
  }
}

package application;

import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.influxdb.dto.Point.Builder;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;

/**
 * 
 * @author zyg
 *
 */
public class InfluxDBConnect {
  private String username;// 用戶名
  private String password;// 密碼
  private String openurl;// 連接地址
  private String database;// 數(shù)據(jù)庫

  private InfluxDB influxDB;

  public InfluxDBConnect(String username, String password, String openurl, String database) {
    this.username = username;
    this.password = password;
    this.openurl = openurl;
    this.database = database;
  }

  /** 連接時序數(shù)據(jù)庫；獲得InfluxDB **/
  public InfluxDB influxDbBuild() {
    if (influxDB == null) {
      influxDB = InfluxDBFactory.connect(openurl, username, password);
      influxDB.createDatabase(database);

    }
    return influxDB;
  }

  /**
   * 設(shè)置數(shù)據(jù)保存策略 defalut 策略名 /database 數(shù)據(jù)庫名/ 30d 數(shù)據(jù)保存時限30天/ 1 副本個數(shù)為1/ 結(jié)尾DEFAULT
   * 表示 設(shè)為默認(rèn)的策略
   */
  public void createRetentionPolicy() {
    String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
        "defalut", database, "30d", 1);
    this.query(command);
  }

  /**
   * 查詢
   * 
   * @param command
   *      查詢語句
   * @return
   */
  public QueryResult query(String command) {
    return influxDB.query(new Query(command, database));
  }

  /**
   * 插入
   * 
   * @param measurement
   *      表
   * @param tags
   *      標(biāo)簽
   * @param fields
   *      字段
   */
  public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields) {
    Builder builder = Point.measurement(measurement);
    builder.time(((long)fields.get("currentTime"))*1000000, TimeUnit.NANOSECONDS);
    builder.tag(tags);
    builder.fields(fields);
    //
    influxDB.write(database, "", builder.build());
  }

  /**
   * 刪除
   * 
   * @param command
   *      刪除語句
   * @return 返回錯誤信息
   */
  public String deleteMeasurementData(String command) {
    QueryResult result = influxDB.query(new Query(command, database));
    return result.getError();
  }

  /**
   * 創(chuàng)建數(shù)據(jù)庫
   * 
   * @param dbName
   */
  public void createDB(String dbName) {
    influxDB.createDatabase(dbName);
  }

  /**
   * 刪除數(shù)據(jù)庫
   * 
   * @param dbName
   */
  public void deleteDB(String dbName) {
    influxDB.deleteDatabase(dbName);
  }

  public String getUsername() {
    return username;
  }

  public void setUsername(String username) {
    this.username = username;
  }

  public String getPassword() {
    return password;
  }

  public void setPassword(String password) {
    this.password = password;
  }

  public String getOpenurl() {
    return openurl;
  }

  public void setOpenurl(String openurl) {
    this.openurl = openurl;
  }

  public void setDatabase(String database) {
    this.database = database;
  }
}

package application;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 
 * @author zyg
 *
 */
@Configuration
public class InfluxDBConfiguration {
  
  private String username = "admin";//用戶名
  private String password = "admin";//密碼
  private String openurl = "http://IP:8086";//InfluxDB連接地址
  private String database = "test_db";//數(shù)據(jù)庫
  
  @Bean
  public InfluxDBConnect getInfluxDBConnect(){
    InfluxDBConnect influxDB = new InfluxDBConnect(username, password, openurl, database);
    
    influxDB.influxDbBuild();
    
    influxDB.createRetentionPolicy();
    return influxDB;
  }
}

package application;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * 
 * @author zyg
 *
 */
@Component
public class Consumer {
  protected final static Logger logger = LoggerFactory.getLogger(Consumer.class);

  private ObjectMapper objectMapper = new ObjectMapper();

  @Autowired
  private InfluxDBConnect influxDB;

  @RabbitListener(queues = RabbitMQConfig.QUEUE_NAME)
  public void sendToSubject(org.springframework.amqp.core.Message message) {

    String payload = new String(message.getBody());
    logger.info(payload);

    if (payload.startsWith("\"")) {
      // Legacy payload from an Angel client
      payload = payload.substring(1, payload.length() - 1);
      payload = payload.replace("\\\"", "\"");
    }
    try {
      if (payload.startsWith("[")) {
        @SuppressWarnings("unchecked")
        List<Map<String, Object>> list = this.objectMapper.readValue(payload, List.class);
        for (Map<String, Object> map : list) {
          sendMap(map);
        }
      } else {
        @SuppressWarnings("unchecked")
        Map<String, Object> map = this.objectMapper.readValue(payload, Map.class);
        sendMap(map);
      }
    } catch (IOException ex) {
      logger.error("Error receiving hystrix stream payload: " + payload, ex);
    }
  }

  private void sendMap(Map<String, Object> map) {
    Map<String, Object> data = getPayloadData(map);
    data.remove("latencyExecute");
    data.remove("latencyTotal");
    Map<String, String> tags = new HashMap<String, String>();
    
    tags.put("type", data.get("type").toString());
    tags.put("name", data.get("name").toString());
    tags.put("instanceId", data.get("instanceId").toString());
    //tags.put("group", data.get("group").toString());
    
    
    influxDB.insert("testaaa", tags, data);

    // for (String key : data.keySet()) {
    // logger.info("{}:{}",key,data.get(key));
    // }


  }

  public static Map<String, Object> getPayloadData(Map<String, Object> jsonMap) {
    @SuppressWarnings("unchecked")
    Map<String, Object> origin = (Map<String, Object>) jsonMap.get("origin");
    String instanceId = null;
    if (origin.containsKey("id")) {
      instanceId = origin.get("host") + ":" + origin.get("id").toString();
    }
    if (!StringUtils.hasText(instanceId)) {
      // TODO: instanceid template
      instanceId = origin.get("serviceId") + ":" + origin.get("host") + ":" + origin.get("port");
    }
    @SuppressWarnings("unchecked")
    Map<String, Object> data = (Map<String, Object>) jsonMap.get("data");
    data.put("instanceId", instanceId);
    return data;
  }

}

這里不多說，就是接收RabbitMQ信息然后保存到InfluxDB數(shù)據(jù)庫中。

三、JVM相關(guān)數(shù)據(jù)采集

JVM相關(guān)數(shù)據(jù)采集非常簡單主要思想就是定時輪訓(xùn)被監(jiān)控服務(wù)的接口地址然后把返回信息插入到InfluxDB中

服務(wù)引用的包不多說這個服務(wù)是需要注冊到注冊中心Eureka中的因為需要獲取所有服務(wù)的監(jiān)控信息。

插入InfluxDB代碼和上面基本類似只不過多了一個批量插入方法

package com.zjs.collection;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;

/**
 * 
 * @author zyg
 *
 */
@EnableEurekaClient
@SpringBootApplication
public class ApplictionCollection 
{
  public static void main(String[] args) {
    SpringApplication.run(ApplictionCollection.class, args);
  }
}

/**
   * 批量插入
   * 
   * @param measurement
   *      表
   * @param tags
   *      標(biāo)簽
   * @param fields
   *      字段
   */
  public void batchinsert(String measurement, Map<String, String> tags, List<Map<String, Object>> fieldslist) {
    org.influxdb.dto.BatchPoints.Builder batchbuilder=BatchPoints.database(database);

    for (Map<String, Object> map : fieldslist) {
      Builder builder = Point.measurement(measurement);
      tags.put("instanceId", map.get("instanceId").toString());
      builder.time((long)map.get("currentTime"), TimeUnit.NANOSECONDS);
      builder.tag(tags);
      builder.fields(map);
      batchbuilder.point(builder.build());
    }
    
    System.out.println(batchbuilder.build().toString());
    
    influxDB.write(batchbuilder.build());
  }

package com.zjs.collection;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.annotation.Bean;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

/**
 * 獲取微服務(wù)實例
 * 
 * @author zyg
 *
 */
@Component
@SpringBootApplication
@EnableScheduling
public class MicServerInstanceInfoHandle {

  protected final static Logger logger = LoggerFactory.getLogger(MicServerInstanceInfoHandle.class);

  final String pathtail = "/metrics/mem.*|heap.*|threads.*|gc.*|nonheap.*|classes.*";

  Map<String, String> tags;

  ThreadPoolExecutor threadpool;

  @Autowired
  DiscoveryClient dc;

  @Autowired
  RestTemplate restTemplate;

  final static LinkedBlockingQueue<Map<String, Object>> jsonMetrics = new LinkedBlockingQueue<>(1000);

  /**
   * 初始化實例 可以吧相關(guān)參數(shù)設(shè)置到配置文件
   */
  public MicServerInstanceInfoHandle() {
    
    tags = new HashMap<String, String>();
    threadpool = new ThreadPoolExecutor(4, 20, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));

  }

  @Autowired
  private InfluxDBConnect influxDB;

  /**
   * metrics數(shù)據(jù)獲取
   */
  @Scheduled(fixedDelay = 2000)
  public void metricsDataObtain() {
    logger.info("開始獲取metrics數(shù)據(jù)");
    List<String> servicelist = dc.getServices();
    for (String str : servicelist) {

      List<ServiceInstance> silist = dc.getInstances(str);

      for (ServiceInstance serviceInstance : silist) {
        threadpool.execute(new MetricsHandle(serviceInstance));
      }
    }
  }

  /**
   * 將數(shù)據(jù)插入到influxdb數(shù)據(jù)庫
   */
  @Scheduled(fixedDelay = 5000)
  public void metricsDataToInfluxDB() {
    logger.info("開始批量將metrics數(shù)據(jù)insert-influxdb");
    ArrayList<Map<String, Object>> metricslist = new ArrayList<>();
    MicServerInstanceInfoHandle.jsonMetrics.drainTo(metricslist);

    if (!metricslist.isEmpty()) {
      logger.info("批量插入條數(shù):{}", metricslist.size());
      influxDB.batchinsert("metrics", tags, metricslist);
    }

    logger.info("結(jié)束批量metrics數(shù)據(jù)insert");
  }

  @Bean
  public RestTemplate getRestTemplate() {
    RestTemplate restTemplate = new RestTemplate();
    SimpleClientHttpRequestFactory achrf = new SimpleClientHttpRequestFactory();
    achrf.setConnectTimeout(10000);
    achrf.setReadTimeout(10000);
    restTemplate.setRequestFactory(achrf);
    return restTemplate;

  }

  class MetricsHandle extends Thread {
    
    private ServiceInstance serviceInstanc;
    
    public MetricsHandle(ServiceInstance serviceInstance){
      serviceInstanc=serviceInstance;
    }
    
    @Override
    public void run() {

      try {
        
        logger.info("獲取 {}:{}:{} 應(yīng)用metrics數(shù)據(jù)",serviceInstanc.getServiceId(),serviceInstanc.getHost(),serviceInstanc.getPort());
        
        @SuppressWarnings("unchecked")
        Map<String, Object> mapdata = restTemplate
            .getForObject(serviceInstanc.getUri().toString() + pathtail, Map.class);
        mapdata.put("instanceId", serviceInstanc.getServiceId() + ":" + serviceInstanc.getHost() + ":"
            + serviceInstanc.getPort());
        mapdata.put("type", "metrics");
        mapdata.put("currentTime", System.currentTimeMillis() * 1000000);
        MicServerInstanceInfoHandle.jsonMetrics.add(mapdata);

      } catch (Exception e) {
        logger.error("instanceId:{},host:{},port:{},path:{},exception:{}", serviceInstanc.getServiceId(),
            serviceInstanc.getHost(), serviceInstanc.getPort(), serviceInstanc.getUri(),
            e.getMessage());
      }
    }
  }

}

這里簡單解釋一下這句代碼 final String pathtail = "/metrics/mem.*|heap.*|threads.*|gc.*|nonheap.*|classes.*"; ，metrics這個路徑下的信息很多但是我們不是都需要所以我們需要有選擇的獲取這樣節(jié)省流量和時間。上面關(guān)鍵類MicServerInstanceInfoHandle做了一個多線程訪問主要應(yīng)對注冊中心有成百上千個服務(wù)的時候單線程可能輪序不過來，同時做了一個隊列緩沖，批量插入到InfluxDB。

四、結(jié)果展示

如果你數(shù)據(jù)采集成功了就可以繪制出來上面的圖形下面是對應(yīng)的sql

SELECT mean("rollingCountFallbackSuccess"), mean("rollingCountSuccess") FROM "testaaa" WHERE ("instanceId" = 'IP:spring-cloud-server1-test:8082' AND "type" = 'HystrixCommand') AND $timeFilter GROUP BY time($__interval) fill(null)

SELECT mean("currentPoolSize") FROM "testaaa" WHERE ("type" = 'HystrixThreadPool' AND "instanceId" = '10.10.12.51:spring-cloud-server1-test:8082') AND $timeFilter GROUP BY time($__interval) fill(null)

SELECT "heap", "heap.committed", "heap.used", "mem", "mem.free", "nonheap", "nonheap.committed", "nonheap.used" FROM "metrics" WHERE ("instanceId" = 'SPRING-CLOUD-SERVER1-TEST:10.10.12.51:8082') AND $timeFilter

好了到這里就基本結(jié)束了。

五、優(yōu)化及設(shè)想

上面的基礎(chǔ)服務(wù)肯定都是需要高可用的，毋庸置疑都是需要學(xué)習(xí)的。如果有時間我也會向大家一一介紹，大家亦可以去搜索相關(guān)資料查看！

可能有人問有一個叫telegraf的小插件直接就能收集相關(guān)數(shù)據(jù)進行聚合結(jié)果監(jiān)控，

其實我之前也是使用的telegraf這個小工具但是發(fā)現(xiàn)一個問題，

就是每次被監(jiān)控的應(yīng)用重啟的時候相關(guān)字段名就會變，

因為他采集使用的是類實例的名字作為字段名，這應(yīng)我們會很不方便，每次重啟應(yīng)用我們都要重新設(shè)置sql語句這樣非常不友好，

再次感覺收集數(shù)據(jù)編碼難度不大所以自己就寫了收集數(shù)據(jù)的代碼！如果有哪位大神對telegraf比較了解可以解決上面我說的問題記得給我留言哦！在這里先感謝！

有些地方是需要優(yōu)化的，比如一些IP端口什么的都是可以放到配置文件里面的。

六、總結(jié)

從spring boot到現(xiàn)在短短的2、3年時間就迅速變得火爆，知識體系也變得完善，開發(fā)成本越來越低，

所以普及程度就越來越高，微服務(wù)雖然很好但是我們也要很好的善于運用，監(jiān)控就是重要的一環(huán)，