|
|
@@ -1,50 +1,53 @@
|
|
|
package main
|
|
|
|
|
|
import (
|
|
|
- "gopkg.in/mgo.v2/bson"
|
|
|
"log"
|
|
|
qu "qfw/util"
|
|
|
+ elastic "qfw/util/elastic"
|
|
|
"sync"
|
|
|
"time"
|
|
|
-)
|
|
|
|
|
|
+ "gopkg.in/mgo.v2/bson"
|
|
|
+)
|
|
|
|
|
|
-func winnerEsTaskOnce() {
|
|
|
+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))
|
|
|
- log.Println("区间id:",task_sid,task_eid)
|
|
|
+ log.Println("区间id:", task_sid, task_eid)
|
|
|
//区间id
|
|
|
q := map[string]interface{}{
|
|
|
"_id": map[string]interface{}{
|
|
|
"$gte": qu.StringTOBsonId(task_sid),
|
|
|
- "$lt": qu.StringTOBsonId(task_eid),
|
|
|
+ "$lt": qu.StringTOBsonId(task_eid),
|
|
|
},
|
|
|
}
|
|
|
//参数
|
|
|
- threadnum:=qu.IntAll(winner_extract["threadnum"])
|
|
|
- db_c1:=qu.ObjToString(winner_extract["db_c1"])
|
|
|
- db_c2:=qu.ObjToString(winner_extract["db_c2"])
|
|
|
- es_index:=qu.ObjToString(winner_extract["es_index"])
|
|
|
- es_type:=qu.ObjToString(winner_extract["es_type"])
|
|
|
-
|
|
|
+ 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 := winnermgo.GetMgoConn()
|
|
|
- defer mgo.DestoryMongoConn(sess)
|
|
|
+ sess := mgostandard.GetMgoConn()
|
|
|
+ defer mgostandard.DestoryMongoConn(sess)
|
|
|
//es
|
|
|
- EsConn := winner_es.GetEsConn()
|
|
|
- defer winner_es.DestoryEsConn(EsConn)
|
|
|
-
|
|
|
- it_1 := sess.DB(winnermgo.DbName).C(db_c1).Find(&q).Sort("_id").Iter()
|
|
|
- num_1:=0
|
|
|
- pool := make(chan bool, threadnum)
|
|
|
- wg := &sync.WaitGroup{}
|
|
|
- for tmp := make(map[string]interface{}); it_1.Next(&tmp);num_1++{
|
|
|
- if num_1%100 == 0 && num_1>0{
|
|
|
- log.Println("当前表:",db_c1,"数量:",num_1)
|
|
|
+ // EsConn := winner_es.GetEsConn()
|
|
|
+ // defer winner_es.DestoryEsConn(EsConn)
|
|
|
+ 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)
|
|
|
@@ -53,25 +56,28 @@ func winnerEsTaskOnce() {
|
|
|
<-pool
|
|
|
wg.Done()
|
|
|
}()
|
|
|
- savetmp := make(map[string]interface{}, 0)
|
|
|
+ savetmp := map[string]interface{}{}
|
|
|
tmp_id := qu.BsonIdToSId(tmp["_id"])
|
|
|
savetmp["_id"] = tmp_id
|
|
|
savetmp["name"] = tmp["company_name"]
|
|
|
savetmp["pici"] = tmp["updatetime"]
|
|
|
- if _, err := EsConn.Index().Index(es_index).Type(es_type).Id(tmp_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
|
|
|
- log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
|
|
|
+ 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{})
|
|
|
}
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- it_2 := sess.DB(winnermgo.DbName).C(db_c2).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("当前表:",db_c2,"数量:",num_1)
|
|
|
+ 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_1)
|
|
|
}
|
|
|
pool <- true
|
|
|
wg.Add(1)
|
|
|
@@ -80,18 +86,30 @@ func winnerEsTaskOnce() {
|
|
|
<-pool
|
|
|
wg.Done()
|
|
|
}()
|
|
|
- savetmp := make(map[string]interface{}, 0)
|
|
|
+ savetmp := map[string]interface{}{}
|
|
|
tmp_id := qu.BsonIdToSId(tmp["_id"])
|
|
|
savetmp["_id"] = tmp_id
|
|
|
savetmp["name"] = tmp["name"]
|
|
|
savetmp["pici"] = tmp["updatetime"]
|
|
|
- if _, err := EsConn.Index().Index(es_index).Type(es_type).Id(tmp_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
|
|
|
- log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
|
|
|
+ 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("总计:",num_1+num_2)
|
|
|
-
|
|
|
-}
|
|
|
+ wg.Wait()
|
|
|
+ winerEsLock.Lock()
|
|
|
+ if len(arrEs) > 0 {
|
|
|
+ tmps := arrEs
|
|
|
+ elastic.BulkSave(index, itype, &tmps, true)
|
|
|
+ arrEs = []map[string]interface{}{}
|
|
|
+ }
|
|
|
+ winerEsLock.Unlock()
|
|
|
+ log.Println("总计:", num_1+num_2)
|
|
|
+}
|
