InfluxDB V2.5 C# Hepler 类及使用实例

using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Linq;
using InfluxDB.Client.Writes;
using SqlSugar;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Text;
using System.Threading.Tasks;

namespace Unit
{
    #region InfluxDB 读写工具类

    /// <summary>
    /// InfluxDB 读写工具类
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class InfluxDBCommon<T>
    {
        static string _baseUri { get; set; }  //地址
        static string _token { get; set; }  //token
        static string _org { get; set; }  //组织
        static string _bucket { get; set; }  //桶
        static InfluxDBClientOptions _options { get; set; }  //桶

        /// <summary>
        /// 构造函数
        /// </summary>
        public InfluxDBCommon()
        {
            _baseUri = ConstModel.influxDBUri;
            _token = ConstModel.influxDBToken;
            _org = ConstModel.influxDBOrg;
            _bucket = ConstModel.influxDBBucket;
            _options = InfluxDBClientOptions.Builder
                          .CreateNew()
                          .Url(_baseUri)
                          .AuthenticateToken(_token)
                          .Org(_org)
                          .Bucket(_bucket)
                          .Build();
        }

        /// <summary>
        /// 写入
        /// </summary>
        /// <param name="point"></param>
        public void WritePoint(PointData point)
        {
            var client = InfluxDBClientFactory.Create(_options);
            var writeApiAsync = client.GetWriteApiAsync();
            writeApiAsync.WritePointAsync(point);
        }

        /// <summary>
        /// 批量写入
        /// </summary>
        /// <param name="points"></param>
        public void WritePoints(List<PointData> points)
        {
            var client = InfluxDBClientFactory.Create(_options);
            var writeApiAsync = client.GetWriteApiAsync();
            writeApiAsync.WritePointsAsync(points);
        }

        /// <summary>
        /// 写入
        /// </summary>
        /// <param name="measurement"></param>
        public void WriteMeasurement(T measurement)
        {
            var client = InfluxDBClientFactory.Create(_options);
            var writeApiAsync = client.GetWriteApiAsync();
            writeApiAsync.WriteMeasurementAsync(measurement);
        }

        /// <summary>
        /// 批量写入
        /// </summary>
        /// <param name="measurements"></param>
        public void WriteMeasurements(List<T> measurements)
        {
            var client = InfluxDBClientFactory.Create(_options);
            var writeApiAsync = client.GetWriteApiAsync();
            writeApiAsync.WriteMeasurementsAsync(measurements);
        }

        /// <summary>
        /// 查询返回 集合
        /// </summary>
        /// <param name="exp">exp 查询条件</param>
        /// <param name="orderByDesc">倒序排序字段，为数据类型时间类型的字段</param>
        /// <returns></returns>
        public List<T> QueryBy<TKey>(Expression<Func<T, bool>> exp, Expression<Func<T, TKey>> orderByDesc)
        {
            using (var client = InfluxDBClientFactory.Create(_options))
            {
                var query = from s in InfluxDBQueryable<T>.Queryable(_bucket, _org, client.GetQueryApiSync())
                            select s;
                return query.Where(exp).OrderByDescending(orderByDesc).ToList();
            }
        }

        /// <summary>
        /// 分页查询
        /// </summary>
        /// <typeparam name="TKey"></typeparam>
        /// <param name="exp">where 表达式</param>
        /// <param name="orderByDesc">排序字段</param>
        /// <param name="skip">当前第几页</param>
        /// <param name="take">每页多少条</param>
        /// <param name="count">返回总行数</param>
        /// <returns>返回分页后结果</returns>
        public List<T> QueryBy<TKey>(Expression<Func<T, bool>> exp, Expression<Func<T, TKey>> orderByDesc, int skip, int take, out int count)
        {
            using (var client = InfluxDBClientFactory.Create(_options))
            {
                var query = from s in InfluxDBQueryable<T>.Queryable(_bucket, _org, client.GetQueryApiSync())
                            select s;
                count = query.Where(exp).Count();
                return query.Where(exp).OrderByDescending(orderByDesc).Skip(skip).Take(take).ToList();
            }
        }

        /// <summary>
        /// 查询所有
        /// </summary>
        /// <returns>返回Measurement 所有数据</returns>
        public List<T> QueryAll()
        {
            using (var client = InfluxDBClientFactory.Create(_options))
            {
                var query = from s in InfluxDBQueryable<T>.Queryable(_bucket, _org, client.GetQueryApiSync())
                            select s;
                return query.ToList();
            }
        }

    }

    #endregion

    #region 使用实例
    public class InfluxDBTest
    {
        public void Test() 
        {
            //声明
            var influx = new InfluxDBCommon<PointDataList>();

            //单个写入
            PointDataList point = new PointDataList() { PointCode = "A0005", PointValue = "1", CreateTime = DateTime.Now, CreateTimeTicks = DateTime.Now.Ticks, Timestamp = DateTime.UtcNow };
            influx.WriteMeasurement(point);

            var pointInfo = PointData
             .Measurement("PointDataList")
             .Tag("pointcode", "DB001")
             .Field("pointvalue", "2#报警信息。。。。。");
            influx.WritePoint(pointInfo);

            //批量写入
            List<PointDataList> Lsts = new List<PointDataList>();
            Lsts.Add(new PointDataList() { PointCode = "A0006", PointValue = "true", CreateTime = DateTime.Now, CreateTimeTicks = DateTime.Now.Ticks, Timestamp = DateTime.UtcNow });
            Lsts.Add(new PointDataList() { PointCode = "A0007", PointValue = "false", CreateTime = DateTime.Now, CreateTimeTicks = DateTime.Now.Ticks, Timestamp = DateTime.UtcNow });
            Lsts.Add(new PointDataList() { PointCode = "A0008", PointValue = "123", CreateTime = DateTime.Now, CreateTimeTicks = DateTime.Now.Ticks, Timestamp = DateTime.UtcNow });
            influx.WriteMeasurements(Lsts);

            //条件查询
            var stim = DateTime.Now.AddHours(-12);
            var etim = DateTime.Now;
            var longs = etim.ToLong();
            var exp = Expressionable.Create<PointDataList>()
                 .And(it => it.PointCode == "A0006")
                 .And(it => it.CreateTimeTicks >= stim.Ticks)
                 .And(it => it.CreateTimeTicks < etim.Ticks)
                 .ToExpression();
            var lst = influx.QueryBy(exp, n => n.Timestamp);

            //查询所有
            var lst2 = influx.QueryAll();

            //根据条件查询分页查询
            int skip = 0;//当前第几页
            int take = 10;//每页多少条
            int count = 0;//返回总行数
            var lst3 = influx.QueryBy(exp, n => n.Timestamp, skip, take, out count);
        }

    /// <summary>
    /// 类似于关系数据库的表名 点位表
    /// </summary>
    [Measurement("PointDataList")]
    public class PointDataList
    {
        /// <summary>
        /// 标签 点位名称
        /// </summary>
        [Column("PointCode", IsTag = true)]
        public string PointCode { get; set; }
        /// <summary>
        /// 普通字段 点位值
        /// </summary>
        [Column("PointValue")]
        public string PointValue { get; set; }
        /// <summary>
        /// 创建时间的Ticks
        /// </summary>
        [Column("CreateTimeTicks")]
        public long CreateTimeTicks { get; set; }
        /// <summary>
        /// 创建时间
        /// </summary>
        [Column("CreateTime")]
        public DateTime CreateTime { get; set; }
        /// <summary>
        /// UTC时间
        /// </summary>
        [Column(IsTimestamp = true)]
        public DateTime Timestamp { get; set; }
    }

    }
    #endregion
}