es_index_data/createEsIndex/winnertask.go

package main

import (
	"esindex/config"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"sync"
	"time"
)

func winnerEsTaskOnce() {
	defer util.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 := mongodb.BsonIdToSId(primitive.NewObjectIDFromTimestamp(preTime))
	task_eid := mongodb.BsonIdToSId(primitive.NewObjectIDFromTimestamp(curTime))
	log.Info("winner 区间id", zap.String("sid", task_sid), zap.String("eid", task_eid))
	//区间id
	q := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gte": mongodb.StringTOBsonId(task_sid),
			"$lt":  mongodb.StringTOBsonId(task_eid),
		},
	}
	//mongo
	sess := MgoQ.GetMgoConn()
	defer MgoQ.DestoryMongoConn(sess)

	it_1 := sess.DB(MgoQ.DbName).C("winner_enterprise").Find(&q).Sort("_id").Iter()
	num_1 := 0
	for tmp := make(map[string]interface{}); it_1.Next(&tmp); num_1++ {
		if num_1%2000 == 0 && num_1 > 0 {
			log.Info("winnerEsTaskOnce current", zap.Int("count", num_1))
		}
		pool <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-pool
				wg.Done()
			}()
			savetmp := map[string]interface{}{}
			tmp_id := mongodb.BsonIdToSId(tmp["_id"])
			savetmp["_id"] = tmp_id
			savetmp["name"] = tmp["company_name"]
			savetmp["winner_name"] = tmp["company_name"]
			savetmp["pici"] = tmp["updatetime"]
			if province := util.ObjToString(tmp["province"]); province != "" {
				savetmp["province"] = province
			}
			if city := util.ObjToString(tmp["city"]); city != "" {
				savetmp["city"] = city
			}
			if text := util.ObjToString(tmp["tag_business"]); text != "" {
				savetmp["tag_business"] = text
			}

			winerEsLock.Lock()
			arrEs = append(arrEs, savetmp)
			if len(arrEs) >= EsBulkSize {
				tmps := arrEs
				Es.BulkSave(config.Conf.DB.Es.IndexWinner, tmps)
				arrEs = []map[string]interface{}{}
			}
			winerEsLock.Unlock()
		}(tmp)
		tmp = make(map[string]interface{})
	}

	wg.Wait()
	winerEsLock.Lock()
	if len(arrEs) > 0 {
		tmps := arrEs
		Es.BulkSave(config.Conf.DB.Es.IndexWinner, tmps)
		arrEs = []map[string]interface{}{}
	}
	winerEsLock.Unlock()
	log.Info("winner  over！", zap.Int("总计", num_1))
}