关闭mongo连接

udp_winner/config.json

@@ -19,7 +19,7 @@
   "mgodb_extract_kf": "extract_kf",
   "mgo_qyk_c": "enterprise_qyxy",
   "mgo_qyk_buyer": "buyer_qyxy",
-  "mgo_qyk_agency": "gency_qyxy",
+  "mgo_qyk_agency": "agency_qyxy",
   "redis": "127.0.0.1:6379",
   "redis_winner_db": "1",
   "redis_buyer_db": "2",

udp_winner/timedTaskAgency.go

@@ -46,7 +46,7 @@ func TaskAgency(mapinfo *map[string]interface{}) {
 	}).Select(bson.M{"agency": 1, "agencytel": 1, "agencyperson": 1, "topscopeclass": 1,
 		"agencyaddr": 1}).Iter()
 	if cursor.Err() != nil {
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		log.Println(cursor.Err())
 		return
 	}
@@ -87,7 +87,7 @@ func TaskAgency(mapinfo *map[string]interface{}) {
 			}
 		}
 
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 
 		//遍历redis
 		if scan := conn.Scan(0, "", 100); scan.Err() != nil {
@@ -319,6 +319,7 @@ func TaskAgency(mapinfo *map[string]interface{}) {
 				}
 			}
 		}
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		log.Println("增量合并执行完成 ok", gtid, lteid, overid)
 	}
 
@@ -332,13 +333,13 @@ func TimedTaskAgency() {
 	t2 := time.NewTimer(time.Second * 5)
 	for range t2.C {
 		Fcconn := FClient.GetMgoConn(86400)
-		defer Fcconn.Close()
 		tmpLast := map[string]interface{}{}
 		if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("agency_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
 			if !iter.Next(&tmpLast) {
 				//临时表无数据
 				log.Println("临时表无数据:")
-				t2.Reset(time.Second * 15)
+				t2.Reset(time.Minute * 5)
+				FClient.DestoryMongoConn(Fcconn)
 				continue
 			} else {
 				log.Println("临时表有数据:", tmpLast)
@@ -351,6 +352,7 @@ func TimedTaskAgency() {
 				if cursor == nil {
 					log.Println("查询失败")
 					t2.Reset(time.Second * 5)
+					FClient.DestoryMongoConn(fconn)
 					continue
 				}
 				//遍历临时表数据，匹配不到原始库存入异常表
@@ -594,9 +596,10 @@ func TimedTaskAgency() {
 						}
 					}
 				}
-				fconn.Close()
+				FClient.DestoryMongoConn(fconn)
 			}
 		}
+		FClient.DestoryMongoConn(Fcconn)
 		t2.Reset(time.Minute)
 	}
 }

udp_winner/timedTaskBuyer.go

@@ -46,7 +46,7 @@ func TaskBuyer(mapinfo *map[string]interface{}) {
 	}).Select(bson.M{"buyer": 1, "buyertel": 1, "buyerperson": 1, "topscopeclass": 1,
 		"buyeraddr": 1,"buyerclass":1}).Iter()
 	if cursor.Err() != nil {
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		log.Println(cursor.Err())
 		return
 	}
@@ -87,7 +87,7 @@ func TaskBuyer(mapinfo *map[string]interface{}) {
 			}
 		}
 
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 
 		//遍历redis
 		if scan := conn.Scan(0, "", 100); scan.Err() != nil {
@@ -357,6 +357,8 @@ func TaskBuyer(mapinfo *map[string]interface{}) {
 				}
 			}
 		}
+		SourceClient.DestoryMongoConn(SourceClientcc)
+
 		log.Println("增量合并执行完成 ok", gtid, lteid, overid)
 	}
 
@@ -370,13 +372,13 @@ func TimedTaskBuyer() {
 	t2 := time.NewTimer(time.Second * 5)
 	for range t2.C {
 		Fcconn := FClient.GetMgoConn(86400)
-		defer Fcconn.Close()
 		tmpLast := map[string]interface{}{}
 		if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("buyer_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
 			if !iter.Next(&tmpLast) {
 				//临时表无数据
 				log.Println("临时表无数据:")
-				t2.Reset(time.Second * 15)
+				t2.Reset(time.Minute * 5)
+				FClient.DestoryMongoConn(Fcconn)
 				continue
 			} else {
 				log.Println("临时表有数据:", tmpLast)
@@ -389,6 +391,7 @@ func TimedTaskBuyer() {
 				if cursor == nil {
 					log.Println("查询失败")
 					t2.Reset(time.Second * 5)
+					FClient.DestoryMongoConn(fconn)
 					continue
 				}
 				//遍历临时表数据，匹配不到原始库存入异常表
@@ -630,9 +633,10 @@ func TimedTaskBuyer() {
 					}
 
 				}
-				fconn.Close()
+				FClient.DestoryMongoConn(fconn)
 			}
 		}
+		FClient.DestoryMongoConn(Fcconn)
 		t2.Reset(time.Minute)
 	}
 }

udp_winner/timedTaskWinner.go

@@ -46,7 +46,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 	}).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1, "topscopeclass": 1, "winneraddr": 1}).Iter()
 
 	if cursor.Err() != nil {
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		log.Println(cursor.Err())
 		return
 	}
@@ -86,7 +86,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 				log.Println(err)
 			}
 		}
-		SourceClientcc.Close()
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		//遍历redis
 		if scan := conn.Scan(0, "", 100); scan.Err() != nil {
 			log.Println(scan.Err())
@@ -110,7 +110,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 					for _, vmap := range rValuesMaps {
 						vmap["_id"] = bson.ObjectIdHex(vmap["_id"].(string))
 						if err = FClient.SaveForOld("winner_new", vmap); err != nil {
-							log.Println("存量 FClient.Save err", err,vmap)
+							log.Println("存量 FClient.Save err", err, vmap)
 						}
 					}
 					//log.Println("get redis id err:定时任务处理", err, tmp)
@@ -231,8 +231,8 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 			if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
 				//redis不存在存到临时表,定时任务处理
 				FClient.DbName = Config["mgodb_extract_kf"]
-				if err := FClient.SaveForOld("winner_new", tmp); err!=nil {
-					log.Println("FClient.Save err", err,tmp)
+				if err := FClient.SaveForOld("winner_new", tmp); err != nil {
+					log.Println("FClient.Save err", err, tmp)
 				}
 				//log.Println("get redis id err:定时任务处理", err, tmp)
 				if err := rdb.Close(); err != nil {
@@ -338,6 +338,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 				}
 			}
 		}
+		SourceClient.DestoryMongoConn(SourceClientcc)
 		log.Println("增量合并执行完成 ok", gtid, lteid, overid)
 	}
 
@@ -351,13 +352,13 @@ func TimedTaskWinner() {
 	t2 := time.NewTimer(time.Second * 5)
 	for range t2.C {
 		Fcconn := FClient.GetMgoConn(86400)
-		defer Fcconn.Close()
 		tmpLast := map[string]interface{}{}
 		if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
 			if !iter.Next(&tmpLast) {
 				//临时表无数据
 				log.Println("临时表无数据:")
-				t2.Reset(time.Minute * 1)
+				t2.Reset(time.Minute * 5)
+				FClient.DestoryMongoConn(Fcconn)
 				continue
 			} else {
 				log.Println("临时表有数据:", tmpLast)
@@ -370,6 +371,7 @@ func TimedTaskWinner() {
 				if cursor == nil {
 					log.Println("查询失败")
 					t2.Reset(time.Second * 5)
+					FClient.DestoryMongoConn(fconn)
 					continue
 				}
 				//遍历临时表数据，匹配不到原始库存入异常表
@@ -631,9 +633,10 @@ func TimedTaskWinner() {
 						}
 					}
 				}
-				fconn.Close()
+				FClient.DestoryMongoConn(fconn)
 			}
 		}
+		FClient.DestoryMongoConn(Fcconn)
 		t2.Reset(time.Minute)
 	}
 }