springboot+RabbitMQ+InfluxDB+Grafara监控实践

本文需要有相关spring boot 或spring cloud 相关微服务框架的基础，如果您具备相关基础可以很容易的实现下述过程!!!!!!! 希望

本文的所说对需要的您有所帮助

从这里我们开始进入闲聊阶段。

大家都知道 spring boot整合了很多很多的第三方框架，我们这里就简单讨论和使用 性能监控和jvm监控相关的东西。其他的本文不讨论虽然有些关联，所以开篇有说需要有相关spring boot框架基础说了这么多废话，下面真正进入主题。

这里首先给大家看下整体的数据流程图，其中两条主线一条是接口或方法性能监控数据收集，还有一条是spring boot 微服务jvm相关指标数据采集，最后都汇总到influxdb时序数据库中在用数据展示工具grafara进行数据展示或报警。

〇、基础服务

基础服务比较多，其中包括rabbitmq，eureka注册中心，influxdb，grafara（不知道这些东西 请百度或谷歌一下了解相关知识），下面简单说下各基础服务的功能：

rabbitmq 一款很流行的消息中间件，主要用它来收集spring boot应用监控性能相关信息，为什么是rabbitmq而不是什么别的 kafka等等，因为测试方便性能也够用，spring boot整合的够完善。

eureka 注册中心，一般看过或用过spring cloud相关框架的都知道spring cloud注册中心主要推荐使用eureka！至于为什么不做过多讨论不是本文主要讨论的关注点。本文主要用来同步和获取注册到注册中心的应用的相关信息。

influxdb和grafara为什么选这两个，其他方案如 elasticsearch 、logstash 、kibana，elk的组合等！原因很显然 influxdb是时序数据库数据的压缩比率比其他（elasticsearch ）好的很多（当然本人没有实际测试过都是看一些文档）。同时influxdb使用sql非常类似mysql等关系型数据库入门方便，grafara工具可预警。等等！！！！！！！！！！！

好了工具就简单介绍到这里，至于这些工具怎么部署搭建请搭建先自行找资料学习，还是因为不是本文重点介绍的内容，不深入讨论。如果有docker相关基础的童鞋可以直接下载个镜像启动起来做测试使用（本人就是使用docker启动的上面的基础应用（eureka除外））

一、被监控的应用

这里不多说被监控应用肯定是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>

简单说下呢相关包的功能spring-cloud-starter-netflix-eureka-client用于注册中心使用的包，spring-cloud-starter-stream-rabbit 发送rabbitmq相关包，spring-boot-starter-actuator发布监控相关rest接口包，

spring-cloud-starter-hystrix熔断性能监控相关包。

相关注解

@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

简单解释一下endpoints下面相关配置，主要就是 原来这些路径是需要授权访问的，通过配置让这些路径接口不再是敏感的需要授权访问的接口这应我们就可以轻松的访问注册到注册中心的每个服务的响应的接口。这里插一句接口性能需要在方法上面加上如下类似相关注解，然后才会有相关性能数据输出

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

好了到这里你的应用基本上就具备相关性能输出的能力了。你可以访问

如果是上图的接口 你的应用基本ok，为什么是基本因为你截图没有体现性能信息发送rabbitmq的相关信息。这个需要看日志，加入你失败了评论区在讨论。我们先关注主线。

好的spring boot 应用就先说道这里。开始下一主题

二、性能指标数据采集

刚才访问http://ip:port/hystrix.stream这个显示出来的信息就是借口或方法性能相关信息的输出，如果上面都没有问题的话数据应该发送到了rabbitmq上面了我们直接去rabbitmq上面接收相关数据就可以了。

性能指标数据的采集服务主要应用以下包

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

}

这里不多说，就是接收rabbitmq信息然后保存到influxdb数据库中。

三、jvm相关数据采集

jvm相关数据采集非常简单主要思想就是定时轮训被监控服务的接口地址然后把返回信息插入到influxdb中

服务引用的包不多说这个服务是需要注册到注册中心eureka中的因为需要获取所有服务的监控信息。

插入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
   *      标签
   * @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());
      }
    }
  }

}

这里简单解释一下这句代码 final string pathtail = "/metrics/mem.*|heap.*|threads.*|gc.*|nonheap.*|classes.*"; ，metrics这个路径下的信息很多但是我们不是都需要所以我们需要有选择的获取这样节省流量和时间。上面关键类micserverinstanceinfohandle做了一个多线程访问主要应对注册中心有成百上千个服务的时候单线程可能轮序不过来，同时做了一个队列缓冲，批量插入到influxdb。

四、结果展示

如果你数据采集成功了就可以绘制出来上面的图形下面是对应的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

好了到这里就基本结束了。

五、优化及设想

上面的基础服务肯定都是需要高可用的，毋庸置疑都是需要学习的。如果有时间我也会向大家一一介绍，大家亦可以去搜索相关资料查看！

可能有人问有一个叫telegraf的小插件直接就能收集相关数据进行聚合结果监控，

其实我之前也是使用的telegraf这个小工具但是发现一个问题，

就是每次被监控的应用重启的时候相关字段名就会变，

因为他采集使用的是类实例的名字作为字段名，这应我们会很不方便，每次重启应用我们都要重新设置sql语句这样非常不友好，

再次感觉收集数据编码难度不大所以自己就写了收集数据的代码！如果有哪位大神对telegraf比较了解可以解决上面我说的问题记得给我留言哦！在这里先感谢！

有些地方是需要优化的，比如一些ip端口什么的都是可以放到配置文件里面的。

六、总结

从spring boot到现在短短的2、3年时间就迅速变得火爆，知识体系也变得完善，开发成本越来越低，

所以普及程度就越来越高，微服务虽然很好但是我们也要很好的善于运用，监控就是重要的一环，

试想一下你的机房运行着成千上万的服务，稳定运行和及时发现有问题的服务是多么重要的一件事情！

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。