spring boot 监控处理方案实例详解

<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>

@enablehystrix //开启性能监控

@refreshscope //刷新配置文件 与本章无关

@enableautoconfiguration

@enablefeignclients //rpc调用与本章无关

@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);

  }

}

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

　 @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;

}

<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 = "#" ;

  // 创建队列

  @bean

  public queue queue() {

   return new queue(queue_name);

  }

  // 创建一个 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; // 数据库

  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;

  }

  /** 连接时序数据库；获得influxdb **/

  public influxdb influxdbbuild() {

   if (influxdb == null ) {

    influxdb = influxdbfactory.connect(openurl, username, password);

    influxdb.createdatabase(database);

   }

   return influxdb;

  }

  /**

   * 设置数据保存策略 defalut 策略名 /database 数据库名/ 30d 数据保存时限30天/ 1 副本个数为1/ 结尾default

   * 表示 设为默认的策略

   */

  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

   *   标签

   * @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();

  }

  /**

   * 创建数据库

   *

   * @param dbname

   */

  public void createdb(string dbname) {

   influxdb.createdatabase(dbname);

  }

  /**

   * 删除数据库

   *

   * @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" ; //数据库

  @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;

  }

}

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

   *   标签

   * @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;

/**

  * 获取微服务实例

  *

  * @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 );

  /**

   * 初始化实例 可以吧相关参数设置到配置文件

   */

  public micserverinstanceinfohandle() {

   tags = new hashmap<string, string>();

   threadpool = new threadpoolexecutor( 4 , 20 , 60 , timeunit.seconds, new arrayblockingqueue<>( 100 ));

  }

  @autowired

  private influxdbconnect influxdb;

  /**

   * metrics数据获取

   */

  @scheduled (fixeddelay = 2000 )

  public void metricsdataobtain() {

   logger.info( "开始获取metrics数据" );

   list<string> servicelist = dc.getservices();

   for (string str : servicelist) {

    list<serviceinstance> silist = dc.getinstances(str);

    for (serviceinstance serviceinstance : silist) {

     threadpool.execute( new metricshandle(serviceinstance));

    }

   }

  }

  /**

   * 将数据插入到influxdb数据库

   */

  @scheduled (fixeddelay = 5000 )

  public void metricsdatatoinfluxdb() {

   logger.info( "开始批量将metrics数据insert-influxdb" );

   arraylist<map<string, object>> metricslist = new arraylist<>();

   micserverinstanceinfohandle.jsonmetrics.drainto(metricslist);

   if (!metricslist.isempty()) {

    logger.info( "批量插入条数:{}" , metricslist.size());

    influxdb.batchinsert( "metrics" , tags, metricslist);

   }

   logger.info( "结束批量metrics数据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( "获取 {}:{}:{} 应用metrics数据" ,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());

    }

   }

  }

}

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