优化ES Sender逻辑

entity/format.go

@@ -43,7 +43,7 @@ func FormatServiceWfLog(sourceKey string, message string) (Matedata, error) {
 		}
 
 	}
-	mateItem.Data["timestamp"] = mateItem.create
+	mateItem.Data["timestamp"] = mateItem.create.Format("2006-01-02 15:04:05")
 	result := *mateItem
 	mateItem.reset()
 	MatePool.Put(mateItem)

entity/matedata.go

@@ -44,6 +44,7 @@ func CloseMessageChan() {
 func MatedateSender(ctx context.Context, esClient *elastic.Client) {
 
 	tick := time.NewTicker(3 * time.Second)
+
 	var (
 		SenderMu sync.Mutex
 	)
@@ -52,21 +53,32 @@ func MatedateSender(ctx context.Context, esClient *elastic.Client) {
 	for {
 		select {
 		case m := <-messages:
-
 			indexRequest := elastic.NewBulkIndexRequest().Index(m.Index).Doc(m.Data)
+			SenderMu.Lock()
 			bulkRequest.Add(indexRequest)
+			SenderMu.Unlock()
 
 		case <-tick.C:
 			// Do sends the bulk requests to Elasticsearch
 			SenderMu.Lock()
 			count := bulkRequest.NumberOfActions()
-
 			if count > 0 {
-				log.Printf("Send message to Es: %d : \n", bulkRequest.NumberOfActions())
-				_, err := bulkRequest.Do(ctx)
+				log.Printf("Send messages to Index: %d : \n", bulkRequest.NumberOfActions())
+				response, err := bulkRequest.Do(ctx)
+
 				if err != nil {
+					
 					log.Println("Save Es Error:", err)
 				}
+
+				for _ , v := range response.Items {
+					for _, item := range v{
+						if item.Error != nil {
+							log.Printf("Find Error in ES Result in (%s): %s", item.Index, item.Error.Reason)
+						}
+					}
+				}
+
 				bulkRequest.Reset()
 			}
 			SenderMu.Unlock()

source/customer.go

@@ -79,7 +79,7 @@ func ReadingMessage(ctx context.Context, c *Customer) {
 	// var trycount int
 	// var cstSh, _ = time.LoadLocation("Asia/Shanghai") //上海时区
 	var errMessage strings.Builder
-	// log.Println(c.HandlePipeline.pipe)
+
 
 	var matedata entity.Matedata
 	for {