添加bidding和project 半小时内数据监听

createEsIndex/biddingall.toml

@@ -59,3 +59,8 @@ num = 50
     coll = "bidding"
     gtid = "63b05c800000000000000000"
     lteid = "644e90800000000000000000" ##  2023.5.1  18038591
+
+    [all.12]
+    coll = "bidding"
+    gtid = "644e90800000000000000000"  ##  2023.5.1
+    lteid = "64b95a000000000000000000" ##  2023.7.21

createEsIndex/main.go

@@ -49,7 +49,10 @@ var (
 	//detailLength = 50000 // es保存detail长度
 	fileLength = 200000 // es保存附件文本长度,大于20万 时做一个日志记录
 	//pscopeLength = 32766 // projectscope长度
-	specialNames = make(map[string]bool, 0) //存储181 凭安提供的特企,爬虫采购单位
+	specialNames            = make(map[string]bool, 0) //存储181 凭安提供的特企,爬虫采购单位
+	responselock            sync.Mutex
+	BiddingLastNodeResponse int64 //上次节点接受数据时间
+	ProjectLastNodeResponse int64 //上次节点接受数据时间
 )
 
 func init() {
@@ -67,6 +70,9 @@ func init() {
 		Port: util.IntAll(config.Conf.Udp.JyPort),
 	}
 
+	BiddingLastNodeResponse = time.Now().Unix()
+	ProjectLastNodeResponse = time.Now().Unix()
+
 	log.Info("init success")
 }
 
@@ -81,6 +87,7 @@ func main() {
 	go SaveProjectEs()
 
 	go SaveBidErr()
+	go LastUdpJob() //监听半小时内有无数据
 
 	UdpClient = udp.UdpClient{Local: config.Conf.Udp.LocPort, BufSize: 1024}
 	UdpClient.Listen(processUdpMsg)
@@ -117,6 +124,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 					biddingTaskById(mapInfo)
 				}()
 			case "bidding":
+				BiddingLastNodeResponse = time.Now().Unix()
 				pool <- true
 				go func() {
 					defer func() {
@@ -150,6 +158,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 					biddingTask(mapInfo)
 				}()
 			case "project":
+				ProjectLastNodeResponse = time.Now().Unix()
 				pool <- true
 				go func() {
 					defer func() {
@@ -477,3 +486,43 @@ func task() {
 
 	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()))
+	}
+}