jy-data-extract/fullproject/src_v1/load_data.go

package main

import (
	"encoding/json"
	"log"
	"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)
	q := map[string]interface{}{
		"lasttime": map[string]interface{}{"$gt": starttime},
	}
	it := sess.DB(MongoTool.DbName).C(p.coll).Find(&q).Iter()
	n := 0
	pool := make(chan *ProjectInfo, 100)
	over := make(chan bool)
	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)) //, 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 <-over:
				return
			}
		}
	}()
	for {
		result := make(map[string]interface{})
		if it.Next(&result) {
			go func(res map[string]interface{}) {
				bys, _ := json.Marshal(result)
				var tmp *ProjectInfo
				_ = json.Unmarshal(bys, &tmp)
				tmpMap := make(map[string]InfoField)
				infoMap := result["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
				pool <- tmp
			}(result)
		} else {
			break
		}
	}
	time.Sleep(2 * time.Second)
	over <- 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++
				if tmp != nil {
					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)

}