es_index_data/createEsIndex/main.go

package main

import (
	util "app.yhyue.com/data_processing/common_utils"
	"app.yhyue.com/data_processing/common_utils/elastic"
	"app.yhyue.com/data_processing/common_utils/log"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"app.yhyue.com/data_processing/common_utils/mysqldb"
	"app.yhyue.com/data_processing/common_utils/udp"
	"encoding/json"
	"esindex/config"
	"esindex/oss"
	"fmt"
	"github.com/robfig/cron"
	"go.uber.org/zap"
	"io/ioutil"
	"net"
	"net/http"
	"sync"
	"time"
)

var (
	MgoB   *mongodb.MongodbSim
	MgoP   *mongodb.MongodbSim
	MgoQ   *mongodb.MongodbSim
	MysqlB *mysqldb.Mysql

	Es, Es1 *elastic.Elastic

	UdpClient  udp.UdpClient
	UdpTaskMap = &sync.Map{}
	JyUdpAddr  *net.UDPAddr

	EsBulkSize        = 100                                       // es批量保存大小
	updateBiddingPool = make(chan []map[string]interface{}, 5000) //更新bingding数据
	updateBiddingSp   = make(chan bool, 5)
	saveEsPool        = make(chan map[string]interface{}, 5000) //保存binding数据到es
	saveEsSp          = make(chan bool, 5)
	saveProjectEsPool = make(chan map[string]interface{}, 5000) //保存project数据到es
	saveProjectSp     = make(chan bool, 5)
	saveEsAllPool     = make(chan map[string]interface{}, 5000) //存储单机版es,爬虫采集判重使用
	saveEsAllSp       = make(chan bool, 5)

	saveErrBidPool = make(chan map[string]interface{}, 5000)
	saveBidSp      = make(chan bool, 5)

	//detailLength = 50000 // es保存detail长度
	fileLength = 200000 // es保存附件文本长度,大于20万 时做一个日志记录
	//pscopeLength = 32766 // projectscope长度

)

func init() {
	config.Init("./common.toml")
	InitLog()
	InitMgo()
	InitEs()
	InitField()
	InitEsBiddingField()
	oss.InitOss()
	verifyESFields() //检测es 字段类型

	JyUdpAddr = &net.UDPAddr{
		IP:   net.ParseIP(config.Conf.Udp.JyAddr),
		Port: util.IntAll(config.Conf.Udp.JyPort),
	}

	log.Info("init success")
}

func main() {

	go checkMapJob() //udp 发送邮件
	go task_index()  //定时同步更新winner_enterprise、buyer_enterprise ES索引;这个功能很少变动，几乎不需要维护

	go UpdateBidding() //更新bidding表数据
	go SaveEsMethod()
	go SaveAllEsMethod()
	go SaveProjectEs()

	go SaveBidErr()

	UdpClient = udp.UdpClient{Local: config.Conf.Udp.LocPort, BufSize: 1024}
	UdpClient.Listen(processUdpMsg)
	log.Info("Udp服务监听", zap.String("port:", config.Conf.Udp.LocPort))

	ch := make(chan bool, 1)
	<-ch
}

var pool = make(chan bool, 20)

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	switch act {
	case udp.OP_TYPE_DATA:
		var mapInfo map[string]interface{}
		err := json.Unmarshal(data, &mapInfo)
		log.Info("processUdpMsg", zap.Any("mapInfo:", mapInfo))
		if err != nil {
			UdpClient.WriteUdp([]byte("err:"+err.Error()), udp.OP_NOOP, ra)
		} else if mapInfo != nil {
			key, _ := mapInfo["key"].(string)
			if key == "" {
				key = "udpok"
			}
			go UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
			tasktype, _ := mapInfo["stype"].(string)
			switch tasktype {
			case "index-by-id": //单个索引
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingTaskById(mapInfo)
				}()
			case "bidding":
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingTask(mapInfo)
				}()
			case "biddingall": //补充存量数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingAllTask(mapInfo)
				}()

			case "bidding_all_data": //根据biddingall配置文件，存量迁移数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingAllDataTask()
				}()
			case "bidding_history":
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingTask(mapInfo)
				}()
			case "project":
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					projectTask(data, mapInfo)
				}()
			case "project_all_data": //存量 projectset 数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					projectAllData()
				}()
			case "biddingdata": //es 单机版，采集判重
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingDataTask(data, mapInfo)
				}()
			case "biddingdelbyextracttype": //根据bidding表extracttype=-1，删除es中重复数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingDelByExtracttype(data, mapInfo)
				}()
			case "buyer_all": //buyer 全量数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					buyerall()
				}()
			default:
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					log.Info("err", zap.Any("mapInfo", mapInfo))
				}()
			}
		}
	case udp.OP_NOOP: //下个节点回应
		ok := string(data)
		if ok != "" {
			log.Info("udp re", zap.String("data:", ok))
			UdpTaskMap.Delete(ok)
		}
	}
}

func task_index() {

	c := cron.New()

	_ = c.AddFunc("0 0 0 * * ?", func() { task_winneres() }) //每天凌晨执行一次winner生索引
	_ = c.AddFunc("0 0 1 * * ?", func() { task_buyeres() })  //每天1点执行一次buyer生索引
	c.Start()
}
func task_winneres() {
	log.Info("定时任务，winneres")
	winnerEsTaskOnce()
}
func task_buyeres() {
	log.Info("定时任务，buyeres")
	buyerOnce()
}

type UdpNode struct {
	data      []byte
	addr      *net.UDPAddr
	timestamp int64
	retry     int
}

//UpdateBidding 更新bidding表数据
func UpdateBidding() {
	arru := make([][]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-updateBiddingPool:
			arru[indexu] = v
			indexu++
			if indexu == 200 {
				updateBiddingSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateBiddingSp
					}()
					MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
				}(arru)
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				updateBiddingSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateBiddingSp
					}()
					MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
				}(arru[:indexu])
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		}
	}
}

//SaveBidErr 记录错误信息，暂时记录 附件过长的
func SaveBidErr() {
	arru := make([]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-saveErrBidPool:
			arru[indexu] = v
			indexu++
			if indexu == 200 {
				saveBidSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBidSp
					}()
					MgoB.SaveBulk("bidding_es_err_record", arru...)
				}(arru)
				arru = make([]map[string]interface{}, 200)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveBidSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveBidSp
					}()
					MgoB.SaveBulk("bidding_es_err_record", arru...)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, 200)
				indexu = 0
			}
		}
	}
}

//SaveEsMethod 保存到es
func SaveEsMethod() {
	arru := make([]map[string]interface{}, EsBulkSize)
	indexu := 0
	for {
		select {
		case v := <-saveEsPool:
			arru[indexu] = v
			indexu++
			if indexu == EsBulkSize {
				saveEsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsSp
					}()
					Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
					if config.Conf.DB.Es.IndexTmp != "" {
						Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru)
					}
				}(arru)
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveEsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsSp
					}()
					Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
					if config.Conf.DB.Es.IndexTmp != "" {
						Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru)
					}
				}(arru[:indexu])
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		}
	}
}

func SaveAllEsMethod() {
	arru := make([]map[string]interface{}, EsBulkSize)
	indexu := 0
	for {
		select {
		case v := <-saveEsAllPool:
			arru[indexu] = v
			indexu++
			if indexu == EsBulkSize {
				saveEsAllSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsAllSp
					}()
					Es1.BulkSave("biddingall", arru)
				}(arru)
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveEsAllSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsAllSp
					}()
					Es1.BulkSave("biddingall", arru)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		}
	}
}

func SaveProjectEs() {
	arru := make([]map[string]interface{}, EsBulkSize)
	indexu := 0
	for {
		select {
		case v := <-saveProjectEsPool:
			arru[indexu] = v
			indexu++
			if indexu == EsBulkSize {
				saveProjectSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProjectSp
					}()
					Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
				}(arru)
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveProjectSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveProjectSp
					}()
					Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		}
	}
}

func checkMapJob() {
	if config.Conf.Mail.Send {
		log.Info("checkMapJob", zap.String("to:", config.Conf.Mail.To))
		for {
			UdpTaskMap.Range(func(k, v interface{}) bool {
				now := time.Now().Unix()
				node, _ := v.(*UdpNode)
				if now-node.timestamp > 120 {
					node.retry++
					if node.retry > 5 {
						UdpTaskMap.Delete(k)
						res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", config.Conf.Mail.Api, config.Conf.Mail.To, "field-sync-send-fail", k.(string)))
						if err == nil {
							defer res.Body.Close()
							read, err := ioutil.ReadAll(res.Body)
							log.Info("send mail ...", zap.String("r:", string(read)), zap.Any("err:", err))
						}
					} else {
						log.Info("udp重发", zap.Any("k:", k))
						UdpClient.WriteUdp(node.data, udp.OP_TYPE_DATA, node.addr)
					}
				} else if now-node.timestamp > 10 {
					log.Info("udp任务超时中..", zap.Any("k:", k))
				}
				return true
			})
			time.Sleep(60 * time.Second)
		}
	}
}

func task() {
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)

	ch := make(chan bool, 10)
	wg := &sync.WaitGroup{}

	query := sess.DB("qfw").C("result_replace_repair_log").Find(nil).Iter()
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if count%1000 == 0 {
			util.Debug("current ---", count)
		}
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			if id := util.ObjToString(tmp["replace_id"]); mongodb.IsObjectIdHex(id) {
				biddingTaskById(map[string]interface{}{"infoid": id, "stype": "bidding"})
			}
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()

	util.Debug("over ---", count)
}