Program.cs
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Writes;
using Influxdb2.Demo.model.db;
using Influxdb2.Demo.model.dto;
namespace Influxdb2.Demo
{
internal class Program
{
static string url = "http://127.0.0.1:8086";
static string token = "kYFVcao_7JqOMr_QHztGi4kpy-y0Z7MoqZVWu4rix-PNpzszWxcVFrMij8MAJWqObcXcLhrZ5D08BZxUQtsVlw=="; // token
static string org = "abc";//用户组
static string bucket = "a1";//数据库
static void Main(string[] args)
{
Test();
Console.ReadLine();
}
static async Task Test()
{
InfluxdbUtil.Init(url, token, org, bucket);
// 记录格式插入(table,tag1=val1,tag2=val2 field1=a,field2=b)
string record = $"Park,ParkNo=JLM01,ParkName=aa Flg=\"A\",Space=12i";
await InfluxdbUtil.WriteRecord(record);
// 数据点格式写入
List<PointData> lstPoint = new List<PointData>();
var pointTag = PointData.Measurement("Park")
.Tag("ParkNo", "JLM02")
.Tag("ParkName", "bb")
.Timestamp(DateTime.UtcNow, WritePrecision.Ns);
lstPoint.Add(pointTag.Field("Flg", "A"));
lstPoint.Add(pointTag.Field("Space", 13));
await InfluxdbUtil.WritePoints(lstPoint);
// 实体格式插入
var model = new Park { ParkNo = "JLM03", ParkName = "cc", Space = 11, Timestamp = DateTime.UtcNow };
await InfluxdbUtil.WriteModel(model);
// linq查询
var lstLinq1 = InfluxdbUtil.GetAll();
var lstLinq2 = InfluxdbUtil.GetBy("", null, null);
var avg1 = lstLinq2?.Average(p => p.Space);
// 语句查询
var lstFlux11 = await InfluxdbUtil.QueryAll<Park>();// model
var lstFlux12 = await InfluxdbUtil.QueryAll<ParkDto>("Park");// dto
//
// 过滤查询：time + where + order
string startTime = "2024-10-29T00:00:00Z";
string endTime = "2024-10-30T00:00:00Z";
string temp2 =
$@"from(bucket: ""{bucket}"")
|> range(start:{startTime}, stop:{endTime})
|> filter(fn: (r) => r[""_measurement""] == ""Park"")
|> pivot(rowKey: [""_time""], columnKey: [""_field""], valueColumn: ""_value"")
|> sort(columns: [""_time""], desc: true)
|> filter(fn: (r) => r[""Space""]>12)";
var lstFlux2 = await InfluxdbUtil.QueryList<Park>(temp2);
// 公式计算：求平均值
string temp1 =
$@"from(bucket: ""{bucket}"")
|> range(start: -3d)
|> filter(fn: (r) => r._measurement == ""Park"" and r[""_field""] == ""Space"")
|> mean()
|> keep(columns: [""_value""])";
var avg2 = await InfluxdbUtil.QueryOne(temp1);
// 语句插入(按天分表)
var dtStart = new DateTime(1970, 1, 1).ToUniversalTime();
var dtNow = DateTime.UtcNow;
string table = "log" + dtNow.ToString("yyyyMMdd");
string key = dtNow.ToString("yyyy-MM-ddTHH:mm:ssZ");
int value = (int)(dtNow - dtStart).TotalSeconds;
string data = $"{table},CarNo=JLM02 Flg=\"A\",Money=5.5,LogTime=\"{key}\",LogTimestamp={value}i";
await InfluxdbUtil.WriteRecord(data);
// 语句查询
var start = (int)(Convert.ToDateTime("2024-10-28 00:00:00").ToUniversalTime() - dtStart).TotalSeconds;
string temp3 =
$@"from(bucket: ""{bucket}"")
|> range(start:0)
|> filter(fn: (r) => r[""_measurement""] == ""{table}"")
|> pivot(rowKey: [""_time""], columnKey: [""_field""], valueColumn: ""_value"")
|> sort(columns: [""_time""], desc: true)
|> filter(fn: (r) => r[""LogTimestamp""]>={start})";
var lstLog = await InfluxdbUtil.QueryList<LogDto>(temp3);
Console.Write("End");
}
}
}
InfluxdbUtil.cs
using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Core.Flux.Domain;
using InfluxDB.Client.Linq;
using InfluxDB.Client.Writes;
using Influxdb2.Client;
using Influxdb2.Demo.model.db;
namespace Influxdb2.Demo
{
public class InfluxdbUtil : IDisposable
{
static InfluxDBClient influxdb;
static string DefaultBucket;
static string DefaultOrg;
static WriteApiAsync WriteApiAsync;
static QueryApi QueryApi;
static QueryApiSync QueryApiSync;// linq
static DateTime dtStart = new DateTime(1970, 1, 1).ToLocalTime();
static string DeafultFlg = "A";
/// <summary>
/// 初始化
/// </summary>
public static void Init(string url, string token, string org, string bucket)
{
influxdb = new InfluxDBClient(url, token);
DefaultOrg = org;
DefaultBucket = bucket;
WriteApiAsync = influxdb.GetWriteApiAsync();
QueryApi = influxdb.GetQueryApi();
QueryApiSync = influxdb.GetQueryApiSync();
}
/// <summary>
/// 记录格式写入
/// </summary>
/// <param name="data"></param>
/// <returns></returns>
public static async Task WriteRecord(string data)
{
try
{
await WriteApiAsync.WriteRecordAsync(data, WritePrecision.Ns, DefaultBucket, DefaultOrg);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
/// <summary>
/// 数据点格式写入(单个)
/// </summary>
/// <returns></returns>
public static async Task WritePoint(PointData point)
{
try
{
await WriteApiAsync.WritePointAsync(point, DefaultBucket, DefaultOrg);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
/// <summary>
/// 数据点格式写入(多个)
/// </summary>
/// <param name="points"></param>
/// <returns></returns>
public static async Task WritePoints(List<PointData> points)
{
try
{
await WriteApiAsync.WritePointsAsync(points, DefaultBucket, DefaultOrg);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
/// <summary>
/// 实体格式写入
/// </summary>
public static async Task WriteModel<T>(T t)
{
try
{
await WriteApiAsync.WriteMeasurementAsync(t, WritePrecision.Ns, DefaultBucket, DefaultOrg);
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
}
/// <summary>
/// Linq查询所有
/// </summary>
public static List<Park> GetAll()
{
try
{
var query = from s in InfluxDBQueryable<Park>.Queryable(DefaultBucket, DefaultOrg, QueryApiSync)
select s;
query = query.Where(x => x.Flg == DeafultFlg);// 增加默认过滤项，解决多查询一条null记录
var datas = query.ToList();
return datas;
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
return null;
}
/// <summary>
/// Linq查询where
/// </summary>
public static List<Park> GetBy(string tag, DateTime? stime, DateTime? etime)
{
try
{
var query = from s in InfluxDBQueryable<Park>.Queryable(DefaultBucket, DefaultOrg, QueryApiSync)
select s;
query = query.Where(x => x.Flg == DeafultFlg);
if (!string.IsNullOrEmpty(tag))
{
query = query.Where(x => x.ParkNo == tag);
}
if (stime.HasValue)
{
var tim = stime.Value;
query = query.Where(x => x.Timestamp >= new DateTime(tim.Year, tim.Month, tim.Day, tim.Hour-8, tim.Minute, tim.Second, DateTimeKind.Utc));
}
if (etime.HasValue)
{
var tim = etime.Value;
query = query.Where(x => x.Timestamp < new DateTime(tim.Year, tim.Month, tim.Day, tim.Hour-8, tim.Minute, tim.Second, DateTimeKind.Utc));
}
var datas = query.ToList();
return datas;
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
return null;
}
/// <summary>
/// Fux语法查询所有
/// </summary>
/// <typeparam name="T"></typeparam>
/// <returns></returns>
public static async Task<List<T>> QueryAll<T>(string men = "")
{
if (string.IsNullOrEmpty(men)) men = typeof(T).Name;
//
// 这里借用Influxdb2.Client的Flux对象辅助生成查询字符串
/*
from(bucket: "a1")
|> range(start: -3d)
|> filter(fn: (r) => r._measurement == "park")
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|> sort(columns: ["_time"], desc: true)
|> limit(n: 10, offset: 0)*/
var flux = Flux
.From(DefaultBucket)
.Range("-3d")
.Filter(FnBody.R.MeasurementEquals($"{men}"))
.Pivot()
.Sort(Columns.Time, desc: true)
.Limit(10)
;
var strflux = flux.ToString();
var lstData = await QueryList<T>(strflux);
return lstData;
}
public static async Task<List<T>> QueryList<T>(string strflux)
{
try
{
var query = await QueryApi.QueryAsync<T>(strflux, DefaultOrg);
return query.ToList();
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
return null;
}
public static async Task<List<FluxTable>> QueryTable(string strflux)
{
try
{
var query = await QueryApi.QueryAsync(strflux, DefaultOrg);
return query.ToList();
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
return null;
}
public static async Task<object> QueryOne(string strflux)
{
try
{
var query = await QueryApi.QueryAsync(strflux, DefaultOrg);
var table = query.FirstOrDefault();
return table == null ? null : table.Records[0].Values["_value"];
}
catch (Exception ex)
{
Console.WriteLine(ex.ToString());
}
return null;
}
public void Dispose()
{
influxdb.Dispose();
}
}
}
标签：string,netcore,Influxdb2,await,static,ex,使用,var,query From： https://www.cnblogs.com/chen1880/p/18514029