springboot整合influxdb做业务埋点监控

程序员文章站 2024-01-08 13:05:34

...

安装influxdb

地址：https://portal.influxdata.com/downloads/

找到对应的系统版本，然后依次执行 wget 和 sudo yum语句：

springboot整合influxdb做业务埋点监控

验证是否安装成功：influx -version

springboot整合influxdb做业务埋点监控

启动influx：systemctl start influxdb.service

链接influx客户端：influx

springboot整合influxdb做业务埋点监控

influx基本操作：

查询数据库：show databases

创建数据库：create database xxx

删除数据库：drop database xxx

进入数据库：use xxx

在InfluxDB当中，并没有表（table）这个概念，取而代之的是measurement，measurement的功能与传统数据库中的表一致，因此我们也可以将measurement称为InfluxDB中的表

InfluxDB中没有显式的新建表的语句，只能通过insert数据的方式来建立新表，其中face为表名称，hostname和facename是tag索引，value为记录值：

insert face,hostname=server01,facename=getXxx value=4

查询表数据语句和mysql差不多：select * from xxx

删除表：drop measurement xxx

查询所有表：show measurements

boot整合influx

maven依赖：

<dependency>
    <groupId>org.influxdb</groupId>
    <artifactId>influxdb-java</artifactId>
    <version>2.15</version>
</dependency>

yml配置：

spring:
  influx:
    url: http://192.168.1.211:8086

实现逻辑：用一个异步任务去执行influx插入操作，不影响程序业务逻辑

influx常量类：

import org.influxdb.dto.Point;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

public class InfluxStatics {

    // 初始化容量为100的队列,point容器
    public static BlockingQueue<Point> queue = new LinkedBlockingDeque<>(100);

    // influx表名称
    public static String MEASUREMENT_FACE = "face";

    // 往queue中添加point
    public static void addPoint(String measurement, Map<String, String> tag, Long value) {
        Point.Builder builder = Point.measurement(measurement);
        builder.tag(tag);
        builder.addField("value", value);
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        try {
            queue.put(builder.build());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

influx操作线程类

import org.influxdb.InfluxDB;
import org.influxdb.dto.Point;
import java.util.concurrent.BlockingQueue;

public class InfluxRecord implements Runnable {

    private InfluxDB influxDb;

    private BlockingQueue<Point> blockingQueue;

    public InfluxRecord(InfluxDB influxDb, BlockingQueue<Point> blockingQueue) {
        this.influxDb = influxDb;
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        // 循环取point记录到influx
        while (true) {
            try {
                Point point = blockingQueue.take();
                // 设置数据库
                influxDb.setDatabase("record");
                influxDb.write(point);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}

启动线程

import com.ctrip.basebiz.vcservice.web.job.InfluxRecord;
import com.ctrip.basebiz.vcservice.web.job.InfluxStatics;
import org.influxdb.InfluxDB;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.SpringApplicationRunListener;
import org.springframework.context.ConfigurableApplicationContext;

public class ISpringApplicationRunListener implements SpringApplicationRunListener {

    public ISpringApplicationRunListener(SpringApplication application, String[] args) {}

    // 在ApplicationContext刷新启动后,赋值influxDb对象并启动线程
    public void started(ConfigurableApplicationContext context) {
        InfluxDB influxDb = (InfluxDB) context.getBean("influxDb");
        new Thread(new InfluxRecord(influxDb, InfluxStatics.queue)).start();
    }
}

最后，在业务中进行埋点

import com.ctrip.basebiz.vcservice.web.job.InfluxStatics;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import java.util.HashMap;
import java.util.Map;

@RestController
@RequestMapping("/influx")
public class InfluxController {

    @GetMapping("/getAaa")
    @ResponseBody
    public String getAaa() {
        record("server01", "getAaa");
        return "getAaa success";
    }

    @GetMapping("/getBbb")
    @ResponseBody
    public String getBbb() {
        record("server02", "getBbb");
        return "getBbb success";
    }

    /**
     * 记录接口调用时间
     * @param hostName
     * @param faceName
     */
    public void record(String hostName, String faceName) {
        Map<String, String> tags = new HashMap<>();
        tags.put("hostname", hostName);
        tags.put("facename", faceName);
        InfluxStatics.addPoint(InfluxStatics.MEASUREMENT_FACE, tags, 1L);
    }
}

安装grafana，图形监控

地址：https://grafana.com/grafana/download

springboot整合influxdb做业务埋点监控