根据需要实现动态创建超级表，设备表，灵活实现测量参数的名称类型和tags的适配，不用再每个测试改一个程序，实现全自动生成表，入库表
type ZngValue struct {
	DeviceID    string                 `json:"deviceid"`  //设备名
	Metric      string                 `json:"metric"`    //物理量 表名
	Endpoint    string                 `json:"endpoint"`  //IP
	Timestamp   int64                  `json:"timestamp"` //时间戳
	Step        int64                  `json:"step"`      //时间间隔
	ValuesMap   map[string]interface{} `json:"valuesMap"` //数据
	CounterType string                 `json:"counterType"`
	Tags        string                 `json:"tags"`
	TagsMap     map[string]string      `json:"tagsMap"` //保留2种格式，方便后端组件使用
	Extra       string                 `json:"extra"`
}
func (taosd *TDengineDataSource) autogencreatezngvaluesql(sql string, metric ZngValue) (sqlcmd string, err error) {
	var sqlvaluename, sqlvalue string
	sqlbody := sql
	for name, value := range metric.ValuesMap {

		switch vv := value.(type) {
		case string:
			sqlbody = sqlbody + "," + value.(string)
			logger.Debug(vv)

		case float32:
			//vl := value.(float32)
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatFloat(float64(value.(float32)), 'E', -1, 64)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatFloat(float64(value.(float32)), 'E', -1, 64) + ","

			logger.Debug(vv)
		case float64:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatFloat(value.(float64), 'E', -1, 64)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatFloat(float64(value.(float64)), 'E', -1, 64) + ","
			logger.Debug(vv)
		case uint64:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatInt(int64(value.(uint64)), 10)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatInt(int64(value.(uint64)), 10) + ","
			logger.Debug(vv)
		case int64:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatInt(value.(int64), 10)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatInt(int64(value.(int64)), 10) + ","
			logger.Debug(vv)
		case int:
			//sqlbody = sqlbody + "," + name + "=" + strconv.Itoa(value.(int))
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatInt(int64(value.(int)), 10) + ","
			logger.Debug(vv)
		case bool:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatBool(value.(bool))
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatBool(value.(bool)) + ","
		case int8:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatInt(int64(value.(uint64)), 10)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatInt(int64(value.(int8)), 10) + ","
			logger.Debug(vv)
		case int16:
			//sqlbody = sqlbody + "," + name + "=" + strconv.FormatInt(int64(value.(uint64)), 10)
			sqlvaluename += name + ","
			sqlvalue += strconv.FormatInt(int64(value.(int16)), 10) + ","
			logger.Debug(vv)
		case byte:

		default:
			//	var err error
			//err.(string) = "not support"
			//fmt.Printf()err("not support")
			return "", errors.New("not support")
		}
		//var
	}
	sqlvaluename = sqlvaluename[:len(sqlvaluename)-1]
	sqlvalue = sqlvalue[:len(sqlvalue)-1]
	tl := metric.Timestamp
	tls := strconv.FormatInt(tl, 10)

	sqlcmd = sqlbody + sqlvaluename + ") values(" + tls + "000," + sqlvalue + ")\n"
	return sqlcmd, nil
}

func (taosd *TDengineDataSource) serilizeZngTDengine(m ZngValue, tbn string, db *sql.DB) error {
	idx := taosd.TAOShashID([]byte(tbn))
	sqlcmd := " " + tbn

	//tl := m.Timestamp
	//tls := strconv.FormatInt(tl, 10)
	sqlcmd = sqlcmd + "(ts,"
	sqlcmd, err := taosd.autogencreatezngvaluesql(sqlcmd, m)
	if err != nil {
		logger.Debugf("sqlcmd err %v", err)
		return err
	}
	id := idx % taosd.Section.config.insertsqlworkers
	logger.Debugf("Send-->GO %v"+sqlcmd, id)
	taosd.insertbatchChans[id] <- sqlcmd
	return nil
}
func (taosd *TDengineDataSource) autogencreatezngtabelsql(stbname string, metric ZngValue) (sqlcmd string, err error) {
	//var sqlbody string
	sqlbody := " (ts timestamp"
	for valuename, value := range metric.ValuesMap {

		switch value.(type) {
		case string:
			sqlbody = sqlbody + "," + valuename + " BINARY(100)"
		case float32:
			sqlbody = sqlbody + "," + valuename + " FLOAT"
		case float64:
			sqlbody = sqlbody + "," + valuename + " DOUBLE"
		case uint64:
			sqlbody = sqlbody + "," + valuename + " BIGINT"
		case int64:
			sqlbody = sqlbody + "," + valuename + " BIGINT"
		case int:
			sqlbody = sqlbody + "," + valuename + " INT"
		case bool:
			sqlbody = sqlbody + "," + valuename + " BOOL"
		case int8:
			sqlbody = sqlbody + "," + valuename + " TINYINT"
		case int16:
			sqlbody = sqlbody + "," + valuename + " SMALLINT"
		case byte:

		default:
			//	var err error
			//err.(string) = "not support"
			//fmt.Printf()err("not support")
			return "", errors.New("not support")
		}
		//var
	}
	sqlcmd = "create table if not exists " + stbname + sqlbody + ") tags(taghash binary(34)"
	return sqlcmd, nil
}

func (taosd *TDengineDataSource) HandleZngStable(metric ZngValue, db *sql.DB) error {
	taglist := list.New()   // save push in data tags name include endpoint / nid
	tbtaglist := list.New() //save ready add to stable table  tags max num(tagnumlimit)
	tagmap := make(map[string]string)
	tbtagmap := make(map[string]string)
	//m := make(metric, len(metric))
	tagnum := taosd.Section.config.tagnumlimit
	//var hasName bool = false
	var metricsName string
	var tbn string = ""
	var ln string = ""
	taglen := taosd.Section.config.taglen
	var nt nametag
	var sqlcmd string
	//var annotlen int
	//fmt.Println(ts)
	j := 0
	metricsName = metric.Metric
	if metricsName != "" {
		tbn += metricsName
		//hasName = true
	} else {
		info := fmt.Sprintf("no name metric")
		logger.Errorf(info)
		return nil
	}
	if metric.Endpoint != "" {
		tagmap["ip"] = metric.Endpoint
		ln = "ip" //strings.ToLower(string(metric.Endpoint))
		taosd.OrderInsertS(ln, taglist)
		taosd.OrderInsertS(ln, tbtaglist)
		tbtagmap[ln] = "y"
		tbn += strings.ToLower(string(metric.Endpoint))
		j++
	}
	if metric.DeviceID != "" {
		tagmap["deviceid"] = metric.DeviceID
		ln = "deviceid" //strings.ToLower(string(metric.Nid))
		taosd.OrderInsertS(ln, taglist)
		taosd.OrderInsertS(ln, tbtaglist)
		tbtagmap[ln] = "y"
		tbn += strings.ToLower(string(metric.DeviceID))
		j++
	}

	for k, v := range metric.TagsMap {
		//j <= tagnum
		j++
		ln = strings.ToLower(string(k))
		//taosd.OrderInsertS(ln, taglist)
		s := string(v)
		if j <= tagnum {
			tbn += s
			taosd.OrderInsertS(ln, taglist)
			taosd.OrderInsertS(ln, tbtaglist)
			//tbtaglist.PushBack(ln)
			if len(s) > taglen {
				s = s[:taglen]
			}
			tagmap[ln] = s
			tbtagmap[ln] = "y"
		}

	}

	if taosd.Section.config.debugprt == 2 {
		t := metric.Timestamp
		//var ns int64 = 0
		//if t/1000000000 > 10 {
		//	tm := t / 1000
		//	ns = t - tm*1000
		//}
		logger.Debugf(" Ts: %v, value: %v, ", time.Unix(t, 0), metric.ValuesMap)
		//logger.Debug(ts)
	}

	stbname := taosd.tablenameEscape(metricsName)
	//var ok bool
	//i := 0
	schema, ok := taosd.IsSTableCreated.Load(stbname)
	if !ok { // no local record of super table structure
		//获取表结构的TAG字段数组
		stablehas := false
		tags := taosd.taosdGetTableTagDescribe(db, stbname)
		if tags != nil {
			taostaglist := list.New()
			taostagmap := make(map[string]string)
			for _, tag := range tags {
				taostaglist.PushBack(tag)
				taostagmap[tag] = "y"
			}
			nt.taglist = taostaglist
			nt.tagmap = taostagmap
			//tbtaglist = nt.taglist
			//tbtagmap = nt.tagmap
			stablehas = true //stable is exist

		}
		//nt.tagmap
		//nt.taglist

		if stablehas { //超级表存在
			//需要插入的 tags
			//yes, the super table was already created in TDengine
			for e := tbtaglist.Front(); e != nil; e = e.Next() {
				k := e.Value.(string)
				//i++
				//if i < taosd.Section.config.tagnumlimit {
				_, ok = nt.tagmap[k] //表结构的tags name
				if !ok {
					//tag以前不存在，需要添补插入，tag在超级表里没找到则改变表结构，添加此tag
					sqlcmd = "alter table " + stbname + " add tag " + k + taosd.Section.tagstr + "\n"
					_, err := taosd.execSql(taosd.Section.config.dbName, sqlcmd, db)
					if err != nil {
						logger.Error(err)
						errorcode := fmt.Sprintf("%s", err)
						if strings.Contains(errorcode, "duplicated column names") {
							nt.taglist.PushBack(k)
							//OrderInsertS(k, tbtaglist)
							nt.tagmap[k] = "y"
						}
					} else {
						nt.taglist.PushBack(k)
						//OrderInsertS(k, tbtaglist)
						nt.tagmap[k] = "y"
					}
				}
				//}
			}
			tbtaglist = nt.taglist
			tbtagmap = nt.tagmap
			taosd.IsSTableCreated.Store(stbname, nt)
		} else { //超级表不存在
			// no, the super table haven't been created in TDengine, create it.
			nt.taglist = tbtaglist
			nt.tagmap = tbtagmap
			sqlcmd, err := taosd.autogencreatezngtabelsql(stbname, metric)
			if err != nil {
				logger.Errorf("autogencreatetabelsql%s", err.Error())
				return err
			}
			//sqlcmd = "create table if not exists " + stbname + " (ts timestamp, value double) tags(taghash binary(34)"
			for e := tbtaglist.Front(); e != nil; e = e.Next() {
				sqlcmd = sqlcmd + "," + e.Value.(string) + taosd.Section.tagstr
			}
			sqlcmd = sqlcmd + ")\n"
			if taosd.Section.config.debugprt == 2 {
				fmt.Printf("SQL:%s", sqlcmd)
			}
			_, err = taosd.execSql(taosd.Section.config.dbName, sqlcmd, db)
			if err == nil {
				//tbtaglist = nt.taglist
				//tbtagmap = nt.tagmap
				taosd.IsSTableCreated.Store(stbname, nt)
			} else {
				logger.Error(err)
				return err
			}
		}
	} else { //有本地 tag信息，就让需要插入的tags跟本地的tag信息进行比对
		ntag := schema.(nametag)
		//tbtaglist = ntag.taglist
		//tbtagmap = ntag.tagmap
		//i := 0
		for e := tbtaglist.Front(); e != nil; e = e.Next() {
			k := e.Value.(string)
			//i++
			//if i < taosd.Section.config.tagnumlimit {
			_, ok := ntag.tagmap[k]
			if !ok {
				sqlcmd = "alter table " + stbname + " add tag " + k + taosd.Section.tagstr + "\n"
				_, err := taosd.execSql(taosd.Section.config.dbName, sqlcmd, db)
				if err != nil {
					logger.Error(err)
					errorcode := fmt.Sprintf("%s", err)
					if strings.Contains(errorcode, "duplicated column names") {
						ntag.taglist.PushBack(k)
						//OrderInsertS(k, tbtaglist)
						ntag.tagmap[k] = "y"
					} else {
						return err
					}

				} else {
					ntag.taglist.PushBack(k)
					//OrderInsertS(k, tbtaglist)
					ntag.tagmap[k] = "y"
				}
			}
			//}
		}

		tbtaglist = ntag.taglist
		tbtagmap = ntag.tagmap
		taosd.IsSTableCreated.Store(stbname, ntag)
	}
	// insert device table data ,tables create auto
	tbnhash := "MD5_" + taosd.md5V2(tbn)
	_, tbcreated := taosd.IsTableCreated.Load(tbnhash)
	//tbtaglist,tbtagmap根据更新过程记录最新的tag结构
	if !tbcreated {
		var sqlcmdhead, sqlcmd string
		sqlcmdhead = "create table if not exists " + tbnhash + " using " + stbname + " tags(\""
		sqlcmd = ""
		i := 0
		for e := tbtaglist.Front(); e != nil; e = e.Next() {
			if e.Value.(string) == "taghash" {
				continue
			}
			tagvalue, has := tagmap[e.Value.(string)]
			if len(tagvalue) > taglen {
				tagvalue = tagvalue[:taglen]
			}

			if i == 0 {
				if has {
					sqlcmd = sqlcmd + "\"" + tagvalue + "\""
				} else {
					sqlcmd = sqlcmd + "null"
				}
				i++
			} else {
				if has {
					sqlcmd = sqlcmd + ",\"" + tagvalue + "\""
				} else {
					sqlcmd = sqlcmd + ",null"
				}
			}

		}

		var keys []string
		var tagHash = ""
		for t := range tagmap {
			keys = append(keys, t)
		}
		sort.Strings(keys)
		for _, k := range keys {
			tagHash += tagmap[k]
		}

		sqlcmd = sqlcmd + ")\n"
		sqlcmd = sqlcmdhead + taosd.md5V2(tagHash) + "\"," + sqlcmd
		_, err := taosd.execSql(taosd.Section.config.dbName, sqlcmd, db)
		if err == nil {
			taosd.IsTableCreated.Store(tbnhash, true)
		} else {
			return err
		}
	}
	if taosd.Section.config.debugprt == 2 {
		fmt.Println("stable:", metric)
	}
	taosd.serilizeZngTDengine(metric, tbnhash, db)
	return nil
}
func (taosd *TDengineDataSource) PushZng2Queue(items []*ZngValue) {
	errCnt := 0
	var req ZngValue
	for _, item := range items {
		req = *item
		var address string

		//取Endpoint或Nid作为唯一标识经过和计算后为通道索引键值
		if item.Endpoint != "" {
			address = item.Endpoint
		} else {
			if item.DeviceID != "" {
				address = item.DeviceID
			} else {
				continue
			}
		}
		if address != "" {
			//stats.Counter.Set("points.in", 1)
			logger.Debugf("taosd Pushzng2Queue :%s", address)
			if taosd.Section.config.debugprt == 2 {
				fmt.Println("taosd Pushzng2Queue:", req)
			}

			idx := taosd.TAOShashID([]byte(address))
			taosd.inputnodeChans[idx%taosd.Section.config.inputworkers] <- req //将需入库参数组写入通道队列
		} else {
			errCnt += 1
		}
	}

}
//接口实例
func (e Deviceinfo) Heart(c *gin.Context) {
	var values taosd.ZngValue
	var items []*taosd.ZngValue
	err := c.ShouldBindBodyWith(&values, binding.JSON)
	if err != nil {
		fmt.Println("api heart", "ShouldBindBodyWith error", err)
		e.Error(500, err, fmt.Sprintf("设备实时状态更新 失败，\r\n失败信息 %s", err.Error()))
		//response.ReturnJSON(c, http.StatusOK, statuscode.InvalidParam.Code,statuscode.InvalidParam.Msg, nil)
		return
	}
	fmt.Println("api heart values ", values)

	//for _, item := range values {
	items = append(items, &values)
	//}
	fmt.Println("/r/napi heart", items)
	if items != nil {
		taosd.Push2Queue(items)
		//e.OK(values.DeviceID, "设备实时状态更新成功")
	} else {
		e.Error(500, err, fmt.Sprintf("设备实时状态更新 失败，\r\n失败信息 %s", err.Error()))

	}
}
//http 接口的 body
{
    "deviceid": "1002",
    "metric": "heart",
    "endpoint": "192.168.0.11",
    "timestamp": 1629037578, 
    "valuesMap":{
        "cputemp": 65,
        "cpuload": 78,
        "gputemp": 76,
        "gpuload": 66,
        "memempty": 30,
        "diskempty": 50
    },
    "tagsMap": {
        "mac": "ae:23:55:ed:66",
        "localaddr": "北京",
        "devicename": "智能设备1"
    }
}