es_index_data/createEsIndex/main.go

package main

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

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

	Es, Es1, Es2 *elastic.Elastic
	PreEs        = make(map[string]*elastic.Elastic, 0) //预处理 索引客户端

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

	EsBulkSize        = 50                                        // 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长度
	specialNames            = make(map[string]bool, 0) //存储181 凭安提供的特企,爬虫采购单位
	responselock            sync.Mutex
	BiddingLastNodeResponse int64 //上次节点接受数据时间
	ProjectLastNodeResponse int64 //上次节点接受数据时间
)

func init() {
	config.InitConf("./common.toml")
	InitLog()
	InitMysql()
	InitMgo()
	InitEs()
	InitField()

	if config.Conf.Env.OpenPre {
		InitPreProcessField()
		InitPreEsClient()
	}

	InitEsBiddingField()
	oss.InitOss()
	verifyESFields() //检测es 字段类型

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

	if config.Conf.Udp.NeAddr != "" {
		NeUdpAddr = &net.UDPAddr{
			IP:   net.ParseIP(config.Conf.Udp.NeAddr),
			Port: util.IntAll(config.Conf.Udp.NePort),
		}
	}

	BiddingLastNodeResponse = time.Now().Unix()
	ProjectLastNodeResponse = time.Now().Unix()

	log.Info("init success")
}

func main() {
	//正式环境才执行定时任务
	if config.Conf.Env.Stype == 0 {
		go LastUdpJob()  //监听半小时内有无数据
		go checkMapJob() //udp 发送邮件
		go task_index()  //定时同步更新winner_enterprise、buyer_enterprise ES索引;这个功能很少变动，几乎不需要维护
	}

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

	go SaveBidErr()
	//添加预处理函数
	if config.Conf.Env.OpenPre {
		go SavePreEsMethod()
		go dealPreProcess()
	}

	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)
			//有udp 转发时
			if config.Conf.Udp.NeAddr != "" {
				go SendUdpMsg(mapInfo, NeUdpAddr)
			}
			tasktype, _ := mapInfo["stype"].(string)
			switch tasktype {
			case "index-by-id": //单个索引,更新pici
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingTaskById(mapInfo)
				}()
			case "index_by_id": //单个索引,不更新pici
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					biddingTaskById(mapInfo)
				}()
			case "bidding":
				BiddingLastNodeResponse = time.Now().Unix()
				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":
				ProjectLastNodeResponse = time.Now().Unix()
				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_once": // 采购单位昨天增量数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					buyerOnce()
				}()
			case "winner_once": // 中标单位昨天增量数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					winnerEsTaskOnce()
				}()
			case "winner_all": // 中标单位存量数据
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					winnerEsAll()
				}()
			case "attachment": // 补充附件采集，对应bidding为bidding_downloadfile_log
				pool <- true
				go func() {
					defer func() {
						<-pool
					}()
					//有单独配置其他操作
					if len(config.Conf.Others) > 0 {
						if v, ok := config.Conf.Others[tasktype]; ok {
							attachmentBiddingTask(mapInfo, v)
						}
					}
				}()
			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 != "" {
						if config.Conf.DB.Es.Addr2 != "" {
							Es2.BulkSave(config.Conf.DB.Es.IndexTmp, arru) // 新集群
						}
						Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru) //老集群
					}
					if config.Conf.DB.Es.Addr2 != "" {
						Es2.BulkSave(config.Conf.DB.Es.Indexb2, 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 != "" {
						if config.Conf.DB.Es.Addr2 != "" {
							Es2.BulkSave(config.Conf.DB.Es.IndexTmp, arru) // 新集群
						}
						Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru)
					}
					if config.Conf.DB.Es.Addr2 != "" {
						Es2.BulkSave(config.Conf.DB.Es.Indexb2, 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)
}

//LastUdpJob 处理UDP 没有接受数据
func LastUdpJob() {
	for {
		responselock.Lock()
		if time.Now().Unix()-BiddingLastNodeResponse >= 1800 {
			BiddingLastNodeResponse = time.Now().Unix() //重置时间
			sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无bidding据进入 ...相关人员检查..."))
		}
		if time.Now().Unix()-ProjectLastNodeResponse >= 1800 {
			ProjectLastNodeResponse = time.Now().Unix() //重置时间
			sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无project数据进入 ...相关人员检查..."))
		}

		responselock.Unlock()
		time.Sleep(300 * time.Second)
	}
}

//sendErrMailApi 发送邮件
func sendErrMailApi(title, body string) {
	var tomail, api string
	if config.Conf.Mail.Send {
		tomail = config.Conf.Mail.To
		api = config.Conf.Mail.Api
	}

	log.Info("sendErrMailApi", zap.Any(tomail, api))
	res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, tomail, title, body))

	if err == nil {
		defer res.Body.Close()
		read, err := ioutil.ReadAll(res.Body)
		if err != nil {
			log.Info("邮件发送成功", zap.String("read", string(read)))
		}
	} else {
		log.Info("sendErrMailApi", zap.String("邮件发送失败", err.Error()))
	}
}