xiugai

customerdata/src/task.go

@@ -192,7 +192,7 @@ func (c *Customer) GetData(stype string) {
 						//开始处理数据
 						wg.Add(1)
 						ch <- true
-						func(tmpHit *es.SearchHit) {
+						go func(tmpHit *es.SearchHit) {
 							defer func() {
 								<-ch
 								wg.Done()

src/history/task.go

@@ -13,7 +13,7 @@ import (
 
 	"gopkg.in/mgo.v2/bson"
 
-	"go.mongodb.org/mongo-driver/bson/primitive"
+	// "go.mongodb.org/mongo-driver/bson/primitive"
 	es "gopkg.in/olivere/elastic.v1"
 )
 
@@ -140,7 +140,7 @@ func (c *Customer) GetData(stype string) {
 						//开始处理数据
 						wg.Add(1)
 						ch <- true
-						func(tmpHit *es.SearchHit) {
+						go func(tmpHit *es.SearchHit) {
 							defer func() {
 								<-ch
 								wg.Done()
@@ -175,7 +175,6 @@ func (c *Customer) GetData(stype string) {
 														tmp["company_phone"] = phone
 													}
 												}
-
 											}
 										}
 									}
@@ -363,14 +362,17 @@ func (c *Customer) AssembelAndSaveData(historyId string, isDup int) {
 	defer qu.Catch()
 	ch := make(chan bool, 10)
 	wg := &sync.WaitGroup{}
+	var datalock = sync.RWMutex{}
 	n := 0
 	if c.PushModel == 0 && len(c.SaveDataMap) > 0 {
 		for _, tmp := range c.SaveDataMap {
 			wg.Add(1)
 			ch <- true
 			go func(data map[string]interface{}) {
+				datalock.Lock()
 				defer func() {
 					<-ch
+					datalock.Unlock()
 					wg.Done()
 				}()
 				AssembelSave(data, c.IsSearchHosp, c.IsSearchEnps, historyId, c.AppId, isDup)
@@ -388,8 +390,10 @@ func (c *Customer) AssembelAndSaveData(historyId string, isDup int) {
 					wg.Add(1)
 					ch <- true
 					go func(data map[string]interface{}) {
+						datalock.Lock()
 						defer func() {
 							<-ch
+							datalock.Unlock()
 							wg.Done()
 						}()
 						AssembelSave(data, c.IsSearchHosp, c.IsSearchEnps, historyId, c.AppId, isDup)
@@ -407,8 +411,10 @@ func (c *Customer) AssembelAndSaveData(historyId string, isDup int) {
 			wg.Add(1)
 			ch <- true
 			go func(data map[string]interface{}) {
+				datalock.Lock()
 				defer func() {
 					<-ch
+					datalock.Unlock()
 					wg.Done()
 				}()
 				AssembelSave(data, c.IsSearchHosp, c.IsSearchEnps, historyId, c.AppId, isDup)

src/history/util_history.go

@@ -9,7 +9,6 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 	. "util"
 
@@ -492,13 +491,11 @@ func AssembelSave(tmp map[string]interface{}, IsSearchHosp, IsSearchEnps bool, h
 			MgoSave.Save(SaveCollbid, tmp)
 		}
 	}
-	var datalock = sync.RWMutex{}
+
 	if isOk {
 		MgoSaveCache <- tmp
-		datalock.Lock()
 		xlsxArr = append(xlsxArr, tmp)
 		xlsxCount++
-		datalock.Unlock()
 	}
 }