Influxdb之springboot记录数据

最近用到了Influxdb数据库，这是一个优秀的时间序列数据库，我们主要用来记录了些监控日志，然后要做些统计查询。这里做个简单的总结。

 项目基于 springboot2.*，整合 influxdb相当简单，引进依赖，1.7为当前最新版：

<dependency>
    <groupId>com.github.miwurster</groupId>
    <artifactId>spring-data-influxdb</artifactId>
    <version>1.7</version>
</dependency>

 

添加配置：

spring:
  influxdb:
     url: http://localhost:8086
     username: test
     password: test
     database: test
     retention-policy: autogen
     connect-timeout: 10
     read-timeout: 30
     write-timeout: 10

表在写入数据时自动创建，measurement相当于表。代码示例：

public void writeData(LoginLogDO loginLogDO) {
    final Point.Builder builder = Point.measurement("loginLogDO")
            .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
builder.tag("loginUser", loginLogDO.getLoginUser())
            .tag("ip", loginLogDO.getIp())
            .tag("result", loginLogDO.getResult())
            .tag("brower", loginLogDO.getBrower())
            .tag("os", loginLogDO.getOs())
            .addField("count", loginLogDO.getCount());
    final Point point = builder.build();
influxDBTemplate.write(point);
}
如上，就可将数据写入数据库。

分页查询 sql示例：

select * from loginLogDO where time>now()-3d and  time<now()
order by time desc limit 10 offset 20

记住：limit后面是pageSize,offset后面是偏移量，就是起始记录数。是(当前页-1)*pageSize.

 limit {pageSize} offset {offset}

下面也可以， 直接用时间纳秒数，秒后面9个0。

select * from loginLogDO where time>1568172960000000000 and  time<1568179960000000000
order by time desc limit 10 offset 20

下面说统计查询。influxdb这方面比较强大。简单好用。
select count(*) from monitorDO where module='spdemo' and time>now()-10m 
 group by time(1m),resultType limit 10

这样会查出一个列表，以resultType 分组的数据。按时间排序显示。查最近10分钟内，以1分钟为
节点单位的10条统计结果。
后面的limit 10 表示是取最靠近time>now()-10m这个时间的10条统计数据， 
不加，则取所有。
结果如下：
name: monitorDO tags: resultType=success time count_count ---- ----------- 1568182080000000000 0 1568182140000000000 0 1568182200000000000 0 1568182260000000000 0 1568182320000000000 0 1568182380000000000 0 1568182440000000000 0 1568182500000000000 0 1568182560000000000 3 1568182620000000000 0 1568182680000000000 2

时间函数也很灵活强大。 d代表天，h代表小时，m代表分， s代表秒。
now()-1d就表示当前时间的减一天的时间点。
group by time(1m) 表示以一分钟为单位进行数据统计。

最后是分页的示例部分代码：

Query countquery=new Query(countsql,influxDBTemplate.getDatabase());
QueryResult countresult = influxDBTemplate.query(countquery);
int total = 0;
if (countresult.getResults().get(0).getSeries() != null) {
    Object totalObj = countresult.getResults().get(0).getSeries().get(0).getValues().get(0).get(1);
total = totalObj == null ? 0 : ((Double) totalObj).intValue();
}
if (total > 0){
    Query query = new Query(selectsql, influxDBTemplate.getDatabase());
QueryResult result = influxDBTemplate.query(query);
log.debug("result==>" + result);
List<List<Object>> mylist = result.getResults().get(0).getSeries().get(0).getValues();
loginColumns = loginColumns == null ? result.getResults().get(0).getSeries().get(0).getColumns() : loginColumns;
    return new CommonPager<>(
            new PageParameter(pageParameter.getCurrentPage(), pageParameter.getPageSize(), total),
mylist.stream().map(this::toMap).collect(Collectors.toList()));

下面是统计的java主要代码：

Query query = new Query(sql.replace(",resultType", ""), influxDBTemplate.getDatabase());
QueryResult result = influxDBTemplate.query(query);if (result.getResults().get(0).getSeries() == null) {
    return null;
}
List<List<Object>> mylist = result.getResults().get(0).getSeries().get(0).getValues();
List<Map<String, Object>> allList= mylist.stream().map(this::toMonitorMinuteMap).collect(Collectors.toList());
Query rstquery = new Query(sql, influxDBTemplate.getDatabase());
QueryResult opresult = influxDBTemplate.query(rstquery);
List<QueryResult.Series> seriesList=opresult.getResults().get(0).getSeries();
List<Map<String, Object>> errorList=null;
List<Map<String, Object>> successList=null;
for(QueryResult.Series series: seriesList){
    if( series.getTags().get("resultType").equals("success")){
        successList=series.getValues().stream().map(this::toMonitorMinuteMap).collect(Collectors.toList());
}else{
        errorList=series.getValues().stream().map(this::toMonitorMinuteMap).collect(Collectors.toList());
}
}

private  Map<String, Object> toMonitorMinuteMap( List<Object> list) {
    Map<String, Object>  map = new HashMap<>();
map.put("time", ((String)list.get(0)).replace("T", " ").replace("Z", ""));
map.put("value",list.get(1));
    return map;
}

好吧，就先介绍这么多哈。希望对大家有所帮助.

补充配置类：

@Configuration
@EnableConfigurationProperties(InfluxDBProperties.class)
@ConditionalOnProperty(prefix = "spring.influxdb", name = "url")
public class InfluxDbConfiguration {

    @Bean
public InfluxDBConnectionFactory connectionFactory(final InfluxDBProperties influxDBProperties) {
        return new InfluxDBConnectionFactory(influxDBProperties);
}

    @Bean
public InfluxDBTemplate<Point> influxDBTemplate(final InfluxDBConnectionFactory connectionFactory) {
        return new InfluxDBTemplate<>(connectionFactory, new PointConverter());
}

    @Bean
public DefaultInfluxDBTemplate defaultTemplate(final InfluxDBConnectionFactory connectionFactory) {
        return new DefaultInfluxDBTemplate(connectionFactory);
}
}

服务类示例： 

@Service("myInfluxDbService")
public class MyInfluxDbServiceImpl implements MyInfluxDbService {

    private InfluxDBTemplate<Point> influxDBTemplate;

    public InfluxDbServiceImpl (InfluxDBTemplate<Point> influxDBTemplate) {
        this.influxDBTemplate=influxDBTemplate;
}