package main import ( "log" qu "qfw/util" elastic "qfw/util/elastic" "sync" "time" "gopkg.in/mgo.v2/bson" ) func winnerEsTaskOnce() { defer qu.Catch() arrEs := []map[string]interface{}{} winerEsLock := &sync.Mutex{} pool := make(chan bool, 3) wg := &sync.WaitGroup{} now := time.Now() preTime := time.Date(now.Year(), now.Month(), now.Day()-1, now.Hour(), 0, 0, 0, time.Local) curTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local) task_sid := qu.BsonIdToSId(bson.NewObjectIdWithTime(preTime)) task_eid := qu.BsonIdToSId(bson.NewObjectIdWithTime(curTime)) // task_sid = "5e6598f82c27dc56292158da" // task_eid = "5f80c8f89a0c261af872294c" log.Println("winner 区间id:", task_sid, task_eid) //区间id q := map[string]interface{}{ "_id": map[string]interface{}{ "$gte": qu.StringTOBsonId(task_sid), "$lt": qu.StringTOBsonId(task_eid), }, } //参数 winnerent, _ := standard["winnerent"].(map[string]interface{}) win_ent := qu.ObjToString(winnerent["collect1"]) //win_enterr := qu.ObjToString(winnerent["collect2"]) index, _ := winnerent["index"].(string) itype, _ := winnerent["type"].(string) //mongo sess := mgostandard.GetMgoConn() defer mgostandard.DestoryMongoConn(sess) log.Println("q:", q, "db:", mgostandard.DbName, "coll:", win_ent) it_1 := sess.DB(mgostandard.DbName).C(win_ent).Find(&q).Sort("_id").Iter() num_1 := 0 for tmp := make(map[string]interface{}); it_1.Next(&tmp); num_1++ { if num_1%100 == 0 && num_1 > 0 { log.Println("当前表:", win_ent, "数量:", num_1) } pool <- true wg.Add(1) go func(tmp map[string]interface{}) { defer func() { <-pool wg.Done() }() savetmp := map[string]interface{}{} tmp_id := qu.BsonIdToSId(tmp["_id"]) savetmp["_id"] = tmp_id savetmp["name"] = tmp["company_name"] savetmp["winner_name"] = tmp["company_name"] savetmp["pici"] = tmp["updatetime"] if province := qu.ObjToString(tmp["province"]); province != "" { savetmp["province"] = province } if city := qu.ObjToString(tmp["city"]); city != "" { savetmp["city"] = city } winerEsLock.Lock() arrEs = append(arrEs, savetmp) if len(arrEs) >= BulkSize { tmps := arrEs elastic.BulkSave(index, itype, &tmps, true) arrEs = []map[string]interface{}{} } winerEsLock.Unlock() }(tmp) tmp = make(map[string]interface{}) } // log.Println("q:", q, "db:", mgostandard.DbName, "coll:", win_enterr) // it_2 := sess.DB(mgostandard.DbName).C(win_enterr).Find(&q).Sort("_id").Iter() // num_2 := 0 // for tmp := make(map[string]interface{}); it_2.Next(&tmp); num_2++ { // if num_2%100 == 0 && num_2 > 0 { // log.Println("当前表:", win_enterr, "数量:", num_2) // } // pool <- true // wg.Add(1) // go func(tmp map[string]interface{}) { // defer func() { // <-pool // wg.Done() // }() // savetmp := map[string]interface{}{} // tmp_id := qu.BsonIdToSId(tmp["_id"]) // savetmp["_id"] = tmp_id // savetmp["name"] = tmp["name"] // savetmp["winner_name"] = tmp["name"] // savetmp["pici"] = tmp["updatetime"] // winerEsLock.Lock() // arrEs = append(arrEs, savetmp) // if len(arrEs) >= BulkSize { // tmps := arrEs // elastic.BulkSave(index, itype, &tmps, true) // arrEs = []map[string]interface{}{} // } // winerEsLock.Unlock() // }(tmp) // tmp = make(map[string]interface{}) // } wg.Wait() winerEsLock.Lock() if len(arrEs) > 0 { tmps := arrEs elastic.BulkSave(index, itype, &tmps, true) arrEs = []map[string]interface{}{} } winerEsLock.Unlock() log.Println("winnerextract 索引完毕! 总计:", num_1) }