抽取配置更新

src/jy/admin/distribution/distribution.go

@@ -27,9 +27,10 @@ func init() {
 		s_table, _ := c.GetPostForm("s_table")
 		instanceid, _ := c.GetPostForm("instanceid")
 		if s_table != "" { //id划段
-			endate, _ := c.GetPostForm("s_date")
-			log.Println(s_table, endate)
-			ecs.IdsRange(s_table, endate)
+			sdate, _ := c.GetPostForm("s_date")
+			edate, _ := c.GetPostForm("e_date")
+			log.Println(s_table, sdate, edate)
+			ecs.IdsRange(s_table, sdate, edate)
 		} else if instanceid != "" { //实例自动释放时间
 			hours, _ := c.GetPostForm("hours")
 			hour := qu.IntAll(hours)

src/jy/cluster/distributed.go

@@ -21,9 +21,9 @@ import (
 var EscIds map[string][]string //id区间
 
 //根据esc数量实例数量id划段
-func IdsRange(table, endate string) int {
-	start := time.Date(2015, 11, 3, 0, 0, 0, 0, time.Local)
-	end, _ := time.ParseInLocation(qu.Date_Short_Layout, endate, time.Local)
+func IdsRange(table, sdate, edate string) int {
+	start, _ := time.ParseInLocation(qu.Date_Short_Layout, sdate, time.Local)
+	end, _ := time.ParseInLocation(qu.Date_Short_Layout, edate, time.Local)
 	EscIds = map[string][]string{}
 	list, _ := db.Mgo.Find("ecs", `{"Status":"Running"}`, nil, nil, false, -1, -1)
 	ids := RangeIdsByDate(len(*list), start, end)
@@ -73,13 +73,13 @@ func RunEcsTask() int {
 }
 
 //id分段
-func RangeIdsByDate(escnum int, start, edate time.Time) map[string][]string {
+func RangeIdsByDate(escnum int, start, end time.Time) map[string][]string {
 	ids := map[string][]string{}
 	task, _ := db.Mgo.FindById("task", qu.ObjToString(ju.Config["udptaskid"]), nil)
 	log.Println(qu.ObjToString((*task)["s_mgoaddr"]), qu.ObjToString((*task)["s_mgodb"]))
 	DB := db.MgoFactory(2, 3, 120, qu.ObjToString((*task)["s_mgoaddr"]), qu.ObjToString((*task)["s_mgodb"]))
-	total := DB.Count("bidding", `{"comeintime":{"$lt":`+fmt.Sprint(edate.Unix())+`}}`)
-	total_back := DB.Count("bidding_back", `{}`)
+	total := DB.Count("bidding", bson.M{"_id": bson.M{"$gte": bson.NewObjectIdWithTime(start), "$lt": bson.NewObjectIdWithTime(end)}})
+	total_back := DB.Count("bidding_back", bson.M{"_id": bson.M{"$gte": bson.NewObjectIdWithTime(start), "$lt": bson.NewObjectIdWithTime(end)}})
 	total += total_back
 	pagesize := (total + escnum - 1) / escnum
 	log.Printf("total:%d total_back:%d pagesize:%d escnum:%d", total, total_back, pagesize, escnum)
@@ -93,8 +93,8 @@ func RangeIdsByDate(escnum int, start, edate time.Time) map[string][]string {
 		for {
 			tmpsid := bson.NewObjectIdWithTime(start)
 			end := start.Add(4 * time.Hour)
-			if end.Unix() > edate.Unix() {
-				eid = bson.NewObjectIdWithTime(edate)
+			if end.Unix() > end.Unix() {
+				eid = bson.NewObjectIdWithTime(end)
 			} else {
 				eid = bson.NewObjectIdWithTime(end)
 			}

src/jy/extract/extractudp.go

@@ -170,12 +170,23 @@ func ExtractByUdp(sid, eid string, instanceId ...string) {
 		}
 	} else { //普通抽取
 		query := bson.M{"_id": bson.M{"$gte": bson.ObjectIdHex(sid), "$lte": bson.ObjectIdHex(eid)}}
-		list, _ := ext.TaskInfo.FDB.Find(ext.TaskInfo.FromColl, query, nil, Fields, false, -1, -1)
-		for _, v := range *list {
-			//log.Println(v["_id"])
-			j := PreInfo(v)
-			ext.TaskInfo.ProcessPool <- true
-			go ext.ExtractProcess(j)
+		count := ext.TaskInfo.FDB.Count(ext.TaskInfo.FromColl, query)
+		pageNum := (count + PageSize - 1) / PageSize
+		limit := PageSize
+		if count < PageSize {
+			limit = count
+		}
+		for i := 0; i < pageNum; i++ {
+			query = bson.M{"_id": bson.M{"$gte": bson.ObjectIdHex(sid)}}
+			log.Printf("page=%d,query=%v", i+1, query)
+			list, _ := ext.TaskInfo.FDB.Find(ext.TaskInfo.FromColl, query, nil, Fields, false, 0, limit)
+			for _, v := range *list {
+				//log.Println(v["_id"])
+				j := PreInfo(v)
+				ext.TaskInfo.ProcessPool <- true
+				go ext.ExtractProcess(j)
+				sid = qu.BsonIdToSId(v["_id"])
+			}
 		}
 	}
 }

src/web/templates/admin/distribution.html

@@ -132,7 +132,8 @@ $(function () {
                   _tit="id划段"
     				tag=[
     					{label:"表名",s_label:"s_table",placeholder:"信息表名",must:true},
-                        {label:"截止日期",s_label:"s_date",placeholder:"2019-01-01",must:true},
+                        {label:"开始日期",s_label:"s_date",placeholder:"2018-01-01",must:true},
+                        {label:"截止日期",s_label:"e_date",placeholder:"2019-01-01",must:true},
     				]   
                 }else if (n=="releasetime"){
                     _tit="释放设置"