节点通知

udp_ocr_conter/config.json

@@ -14,7 +14,7 @@
   "broadcast_ips": "127.0.0.1;192.168.1.2;192.168.1.3;192.168.1.4",
   "broadcast_port": 1490,
   "http_port": "12345",
-  "cornstr": "0 0/5 * * * ?",
+  "cornstr": "* * * * * ?",
   "corntime_consuming": 300,
   "pernum": 5,
   "time_consuming_limit": 60,

udp_ocr_conter/main.go

@@ -31,12 +31,18 @@ func main() {
 		ips := qu.ObjToString(config.Sysconfig["broadcast_ips"])
 		ipsArr := strings.Split(ips, ";")
 		for _, v := range ipsArr {
-			config.Udpclient.WriteUdp([]byte{}, mu.OP_TYPE_DATA, &net.UDPAddr{
+			log.Println("通知udp来取数据1：",v,config.Sysconfig["broadcast_port"],config.Udpclient.WriteUdp([]byte{}, mu.OP_TYPE_DATA, &net.UDPAddr{
 				IP:   net.ParseIP(v),
 				Port: qu.IntAll(config.Sysconfig["broadcast_port"]),
-			})
+			}))
 		}
 	}
+	for _,v:= range config.CID{
+		log.Println("通知udp来取数据2：",v,config.Sysconfig["broadcast_port"],config.Udpclient.WriteUdp([]byte{}, mu.OP_TYPE_DATA, &net.UDPAddr{
+			IP:   net.ParseIP(v),
+			Port: qu.IntAll(config.Sysconfig["broadcast_port"]),
+		}))
+	}
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(writer http.ResponseWriter, request *http.Request) {
 		http.Redirect(writer, request, "/login", http.StatusFound)
@@ -251,7 +257,7 @@ func reload(ip string) string {
 		return "ip不存在"
 	}
 	config.Sys.Lock()
-	if config.CID[qu.ObjToString((*tmp)["InstanceId"])] == true {
+	if config.CID[qu.ObjToString((*tmp)["InstanceId"])] != "" {
 		delete(config.CID, qu.ObjToString((*tmp)["InstanceId"]))
 	}
 	config.Sys.Unlock()
@@ -281,8 +287,8 @@ func reload(ip string) string {
 				(*escObject)[0]["OcrTaskStatus"] = "successful"
 				mongodb.Update("ocr_ecs", bson.M{"_id": (*escObject)[0]["_id"]}, (*escObject)[0], true, false)
 				config.Sys.Lock()
-				if config.CID[qu.ObjToString((*escObject)[0]["InstanceId"])] == false && qu.ObjToString((*escObject)[0]["InstanceId"]) !=""{
-					config.CID[qu.ObjToString((*escObject)[0]["InstanceId"])] = true
+				if config.CID[qu.ObjToString((*escObject)[0]["InstanceId"])] == "" && qu.ObjToString((*escObject)[0]["InstanceId"]) !=""{
+					config.CID[qu.ObjToString((*escObject)[0]["InstanceId"])] = tmpip
 				}
 				config.Sys.Unlock()
 				log.Println((*escObject)[0]["_id"], tmpip, "部署成功")

udp_ocr_conter/src/config/config.go

@@ -3,18 +3,18 @@ package config
 import (
 	"gopkg.in/mgo.v2/bson"
 	"log"
+	mu "mfw/util"
 	"qfw/util"
 	"qfw/util/mongodb"
 	"strings"
 	"sync"
 	"time"
-	mu "mfw/util"
 )
 
 var Udpclient mu.UdpClient //udp对象
 var Sysconfig map[string]interface{}
 var MgoIP, MgoDB, MgoC, MgoFileFiled, SidField, EidField string
-var CID = map[string]bool{}
+var CID = map[string]string{}
 var Sys sync.RWMutex
 
 
@@ -32,7 +32,8 @@ func init() {
 	}
 	mongodb.InitMongodbPool(util.IntAllDef(Sysconfig["dbsize"], 5), MgoIP, MgoDB)
 
-	ocrescs := mongodb.Find("ocr_ecs", bson.M{}, nil, bson.M{"AutoReleaseTime": 1, "InstanceId": 1}, false, -1, -1)
+	Udpclient = mu.UdpClient{Local: Sysconfig["udpip"].(string) + ":" + Sysconfig["udpport"].(string), BufSize: 1024}
+	ocrescs := mongodb.Find("ocr_ecs", bson.M{}, nil, bson.M{"AutoReleaseTime": 1, "InstanceId": 1,"ip_nw":1}, false, -1, -1)
 	if ocrescs != nil || len(*ocrescs) > 0 {
 		for _, v := range (*ocrescs) {
 			if util.ObjToString(v["AutoReleaseTime"]) != "" {
@@ -45,12 +46,12 @@ func init() {
 					if util.ObjToString(v["InstanceId"]) == ""{
 						continue
 					}
-					CID[ util.ObjToString(v["InstanceId"])] = true
+					CID[ util.ObjToString(v["InstanceId"])] = util.ObjToString(v["ip_nw"])
 					//log.Println("删除过时实例", mongodb.Del("ocr_ecs", bson.M{"_id": v["_id"].(bson.ObjectId)}))
 				}
 			}
 		}
 	}
 	log.Println(CID)
-	Udpclient = mu.UdpClient{Local: Sysconfig["udpip"].(string) + ":" + Sysconfig["udpport"].(string), BufSize: 1024}
+
 }

udp_ocr_conter/src/corntask/task_corn.go

@@ -61,7 +61,7 @@ var Auto = func() {
 				}(tmpIid)
 			}
 			config.Sys.Lock()
-			config.CID = make(map[string]bool)
+			config.CID = make(map[string]string)
 			config.Sys.Unlock()
 		} else {
 			var tmpnum int
@@ -112,8 +112,8 @@ var Auto = func() {
 								(*escObject)[i]["OcrTaskStatus"] = "successful"
 								mongodb.Update("ocr_ecs", bson.M{"_id": (*escObject)[i]["_id"]}, (*escObject)[i], true, false)
 								config.Sys.Lock()
-								if config.CID["InstanceId"] == false {
-									config.CID["InstanceId"] = true
+								if config.CID["InstanceId"] == "" {
+									config.CID["InstanceId"] = tmpip
 								}
 								config.Sys.Unlock()
 								log.Println((*escObject)[i]["_id"], tmpip, "部署成功")
@@ -170,8 +170,8 @@ func DynamicTask() {
 							(*escObject)[i]["OcrTaskStatus"] = "successful"
 							mongodb.Update("ocr_ecs", bson.M{"_id": (*escObject)[i]["_id"]}, (*escObject)[i], true, false)
 							config.Sys.Lock()
-							if config.CID[InstanceId] == false && InstanceId != "" {
-								config.CID[InstanceId] = true
+							if config.CID[InstanceId] == "" && InstanceId != "" {
+								config.CID[InstanceId] = tmpip
 							}
 							config.Sys.Unlock()
 							log.Println((*escObject)[i]["_id"], tmpip, "部署成功")

udp_ocr_conter/src/info/taskinfo.go

@@ -4,7 +4,6 @@ import (
 	"config"
 	"fmt"
 	"gopkg.in/mgo.v2/bson"
-	"log"
 	"qfw/util"
 	"qfw/util/mongodb"
 	"time"
@@ -56,7 +55,7 @@ func QueryInfo() map[string]interface{} {
 
 	since := time.Since(stime)
 	data["executionTime"] = since.String()
-	log.Println(now.Sub(stime).Seconds())
+	//log.Println(now.Sub(stime).Seconds())
 	mtmm := float64(overNum) / now.Sub(stime).Seconds() / float64(3+len(config.CID))
 	//log.Println(overNum ,now.Sub(stime).Seconds(),3+len(config.CID),mtmm)
 	tmpf := float64(nowSumNum) / mtmm / float64(3+len(config.CID))
@@ -64,6 +63,6 @@ func QueryInfo() map[string]interface{} {
 	data["estimatedFinishTime"] = now.Add(time.Second*time.Duration(tmpf)).String()
 	data["esc"] = fmt.Sprint(3 ,"+", len(config.CID))
 	data["resultInfos"] = resultInfos
-	log.Println(resultInfos)
+	//log.Println(resultInfos)
 	return data
 }