更换为UDP, 批量1000个写入一次

service/conf/influxdb_conf.go

@@ -7,11 +7,11 @@ import (
 
 func NewClient() client.Client {
 
-	conf := client.HTTPConfig{
-		Addr: "http://172.20.6.29:8086",
+	conf := client.UDPConfig{
+		Addr: "172.20.6.29:8089",
 	}
 
-	con, err := client.NewHTTPClient(conf)
+	con, err := client.NewUDPClient(conf)
 
 	if err != nil {
 		log.Fatal(err)

service/msg_process.go

@@ -26,16 +26,9 @@ func MsgProcess(msg string) {
 	msgInfluxProcess(traceMsg)
 }
 
-func msgInfluxProcess(traceMsgs []TraceMsg) {
-	c := conf.NewClient()
-	defer func() { _ = c.Close() }()
+var pointSlice = make([]*client.Point, 0, 1000)
 
-	points, err := client.NewBatchPoints(client.BatchPointsConfig{
-		Database: "monitor",
-	})
-	if err != nil {
-		log.Fatal(err)
-	}
+func msgInfluxProcess(traceMsgs []TraceMsg) {
 
 	for _, traceMsg := range traceMsgs {
 		if traceMsg.Kind != "SERVER" {
@@ -47,21 +40,18 @@ func msgInfluxProcess(traceMsgs []TraceMsg) {
 			path = traceMsg.Tags.HttpMethod + " " + traceMsg.Tags.HttpPath
 		}
 
-		//fmt.Println("搞起" ,traceMsg.TraceId)
 		sysName := traceMsg.LocalEndpoint.ServiceName
 
 		fields := make(map[string]interface{})
 		fields["duration"] = traceMsg.Duration / 1000
 		fields["traceId"] = traceMsg.TraceId
-		tags := make(map[string]string,)
+		tags := make(map[string]string, )
 		tags["sys_name"] = sysName
 		tags["path"] = path
 		tags["host"] = traceMsg.LocalEndpoint.Ipv4
 		tags["kind"] = traceMsg.Kind
 		tags["duration_tag"] = strconv.Itoa(traceMsg.Duration)
 
-
-
 		bytes, err := json.Marshal(traceMsg)
 
 		msg := string(bytes)
@@ -74,9 +64,14 @@ func msgInfluxProcess(traceMsgs []TraceMsg) {
 
 		unix := time.Unix(0, traceMsg.Timestamp*1000)
 
+		if len(pointSlice) == 1000 {
+			go batchWrite(pointSlice)
+			pointSlice = make([]*client.Point, 0)
+		}
+
 		point, _ := client.NewPoint("trace_info", tags, fields, unix)
-		points.AddPoint(point)
-		err = c.Write(points)
+		pointSlice = append(pointSlice, point)
+
 		if err != nil {
 			log.Fatal(err, msg)
 		}
@@ -84,6 +79,25 @@ func msgInfluxProcess(traceMsgs []TraceMsg) {
 
 }
 
+func batchWrite(pointArray []*client.Point) {
+	c := conf.NewClient()
+	defer func() { _ = c.Close() }()
+
+	points, err := client.NewBatchPoints(client.BatchPointsConfig{
+		Database: "monitor",
+	})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	points.AddPoints(pointArray)
+	err = c.Write(points)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+}
+
 func msgRedisProcess(traceMsg []TraceMsg) {
 	conn := conf.Pool.Get()
 	defer func() { _ = conn.Close() }()