package main import ( "encoding/json" "log" "qfw/util" "qfw/util/redis" "time" ) // 初始加载数据,默认加载最近6个月的数据 func (p *ProjectTask) loadData(starttime int64) { log.Println("load project start..", starttime) p.findLock.Lock() defer p.findLock.Unlock() p.AllIdsMapLock.Lock() defer p.AllIdsMapLock.Unlock() sess := MongoTool.GetMgoConn() defer MongoTool.DestoryMongoConn(sess) loadOver := make(chan bool) q := map[string]interface{}{ "lasttime": map[string]interface{}{"$gte": starttime}, } //field := map[string]interface{}{"list": 0} it := sess.DB(MongoTool.DbName).C(p.coll).Find(&q).Select(nil).Iter() n := 0 pool := make(chan *ProjectCache, 1000) go func() { for { select { case tmp := <-pool: n++ if n%10000 == 0 { log.Println("current", n, "\n", tmp.Id, len(p.mapPn), len(p.mapPc), len(p.mapPb), len(p.mapHref)) //, tmp.ProjectName, tmp.MPN, tmp.ProjectCode, tmp.MPC, tmp.Buyer) } if tmp != nil { id := tmp.Id.Hex() for _, v := range append([]string{tmp.ProjectName}, tmp.MPN...) { if v != "" { //v = pcReplace.ReplaceAllString(v, "") if v != "" { k := p.mapPn[v] if k == nil { k = &Key{Arr: []string{id}} p.mapPn[v] = k } else { k.Arr = append(k.Arr, id) } } } } for _, v := range append([]string{tmp.ProjectCode}, tmp.MPC...) { if v != "" { //v = pcReplace.ReplaceAllString(v, "") if v != "" { k := p.mapPc[v] if k == nil { k = &Key{Arr: []string{id}} p.mapPc[v] = k } else { k.Arr = append(k.Arr, id) } } } } if tmp.Buyer != "" && len([]rune(tmp.Buyer)) > 2 { k := p.mapPb[tmp.Buyer] if k == nil { k = &Key{Arr: []string{id}} p.mapPb[tmp.Buyer] = k } else { k.Arr = append(k.Arr, id) } } p.AllIdsMap[id] = &ID{Id: id, P: tmp} } case <-loadOver: return } } }() for { result := make(map[string]interface{}) if it.Next(&result) { go func(res map[string]interface{}) { bys, _ := json.Marshal(result) var tmp *ProjectCache _ = json.Unmarshal(bys, &tmp) saveFiled(p, result, tmp) redis.PutCKV("project", tmp.Id.Hex(), result) pool <- tmp }(result) } else { break } } time.Sleep(2 * time.Second) loadOver <- true log.Println("load project over..", n) } func (p *ProjectTask) loadSite() { log.Println("load site start..") p.findLock.Lock() defer p.findLock.Unlock() p.mapSiteLock.Lock() defer p.mapSiteLock.Unlock() sess := MongoTool.GetMgoConn() defer MongoTool.DestoryMongoConn(sess) q := map[string]interface{}{} it := sess.DB(MongoTool.DbName).C(SiteColl).Find(&q).Iter() n := 0 pool := make(chan *Site, 100) over := make(chan bool) go func() { for { select { case tmp := <-pool: n++ //站点有效标记state if tmp != nil && tmp.Status == 5 { p.mapSite[tmp.Site] = tmp } case <-over: return } } }() for { result := make(map[string]interface{}) if it.Next(&result) { go func(res map[string]interface{}) { bys, _ := json.Marshal(result) var tmp *Site _ = json.Unmarshal(bys, &tmp) pool <- tmp }(result) } else { break } } time.Sleep(2 * time.Second) over <- true log.Println("load site over..", n) } func saveFiled(p *ProjectTask, res map[string]interface{}, tmp *ProjectCache) { proHref := util.ObjToString(res["projecthref"]) if proHref != "" { p.mapHrefLock.Lock() p.mapHref[proHref] = tmp.Id.Hex() p.mapHrefLock.Unlock() } //tmpMap := make(map[string]InfoField) //infoMap := res["infofield"].(map[string]interface{}) //for _, v := range infoMap { // var field InfoField // b, _ := json.Marshal(v) // _ = json.Unmarshal(b, &field) // tmpMap[tmp.Id.Hex()] = field //} //tmp.InfoFiled = tmpMap } // 加载spidercode数据,isflow字段 func (p *ProjectTask) loadSpiderCode() { log.Println("load spider code start..") p.findLock.Lock() defer p.findLock.Unlock() p.mapSpiderLock.Lock() defer p.mapSpiderLock.Unlock() sess := MgoSpider.GetMgoConn() defer MgoSpider.DestoryMongoConn(sess) q := map[string]interface{}{} field := map[string]interface{}{"code": 1, "isflow": 1} it := sess.DB(MgoSpider.DbName).C("luaconfig").Find(&q).Select(field).Iter() n := 0 pool := make(chan map[string]interface{}, 100) over := make(chan bool) go func() { for { select { case tmp := <-pool: n++ code := util.ObjToString(tmp["code"]) p.mapSpider[code] = util.IntAll(tmp["isflow"]) case <-over: return } } }() for { result := make(map[string]interface{}) if it.Next(&result) { go func(res map[string]interface{}) { pool <- result }(result) } else { break } } time.Sleep(2 * time.Second) over <- true log.Println("load spider over..", n) }