批量保存

src/jy/extract/extractInit.go

@@ -81,12 +81,12 @@ type ExtractTask struct {
 	ResultArr [][]map[string]interface{} //抽取结果详情
 	BidChanel chan bool                  //抽取结果
 	BidArr    [][]map[string]interface{} //抽取结果
-	BidTotal int                         //结果数量
+	BidTotal  int                        //结果数量
 
 	RecogFieldMap map[string]map[string]interface{}   //识别字段
 	FidClassMap   map[string][]map[string]interface{} //分类
-	CidRuleMap map[string][]map[string]interface{}    //规则
-	AuditFields []string                              //需要审核的字段名称
+	CidRuleMap    map[string][]map[string]interface{} //规则
+	AuditFields   []string                            //需要审核的字段名称
 
 	ProvinceMap       map[string]string    //省全称简称(key:浙江省 val:浙江)
 	ProvinceBriefMap  map[string]*Province //省简称对应的省信息(key:浙江 val:&Province{})
@@ -818,9 +818,9 @@ func (e *ExtractTask) ResultSave(init bool) {
 		go func() {
 			for {
 				e.RWMutex.Lock()
-				if len(e.ResultArr) > 50 {
-					arr := e.ResultArr[:50]
-					e.ResultArr = e.ResultArr[50:]
+				if len(e.ResultArr) > saveLimit {
+					arr := e.ResultArr[:saveLimit]
+					e.ResultArr = e.ResultArr[saveLimit:]
 					e.RWMutex.Unlock()
 					qu.Try(func() {
 						db.Mgo.UpSertBulk("extract_result", arr...)
@@ -849,9 +849,9 @@ func (e *ExtractTask) ResultSave(init bool) {
 		qu.Try(func() {
 			lenarr := len(arr)
 			for {
-				if lenarr > 50 {
-					arr2 := arr[:50]
-					arr = arr[50:]
+				if lenarr > saveLimit {
+					arr2 := arr[:saveLimit]
+					arr = arr[saveLimit:]
 					lenarr = len(arr)
 					e.TaskInfo.TDB.UpSertBulk(e.TaskInfo.ToColl, arr2...)
 				} else {
@@ -878,9 +878,9 @@ func (e *ExtractTask) BidSave(init bool) {
 		go func() {
 			for {
 				e.RWMutex.Lock()
-				if len(e.BidArr) > 50 {
-					arr := e.BidArr[:50]
-					e.BidArr = e.BidArr[50:]
+				if len(e.BidArr) > saveLimit {
+					arr := e.BidArr[:saveLimit]
+					e.BidArr = e.BidArr[saveLimit:]
 					e.RWMutex.Unlock()
 					qu.Try(func() {
 						e.TaskInfo.TDB.UpSertBulk(e.TaskInfo.ToColl, arr...)
@@ -897,7 +897,7 @@ func (e *ExtractTask) BidSave(init bool) {
 						log.Debug(err)
 					})
 				}
-				time.Sleep(3 * time.Second)
+				time.Sleep(2 * time.Second)
 			}
 		}()
 	} else {
@@ -908,9 +908,9 @@ func (e *ExtractTask) BidSave(init bool) {
 		qu.Try(func() {
 			lenarr := len(arr)
 			for {
-				if lenarr > 50 {
-					arr2 := arr[:50]
-					arr = arr[50:]
+				if lenarr > saveLimit {
+					arr2 := arr[:saveLimit]
+					arr = arr[saveLimit:]
 					lenarr = len(arr)
 					e.TaskInfo.TDB.UpSertBulk(e.TaskInfo.ToColl, arr2...)
 				} else {
@@ -993,7 +993,7 @@ func (e *ExtractTask) InitAuditRule() {
 func (e *ExtractTask) InitAuditFields() {
 	if len(e.AuditFields) == 0 {
 		v, _ := db.Mgo.FindOne("version", `{"isuse":true,"delete":false}`) //查找当前使用版本
-		if v != nil && len(*v) > 0 { //查找当前使用版本中属性配置需要审核的字段
+		if v != nil && len(*v) > 0 {                                       //查找当前使用版本中属性配置需要审核的字段
 			vid := qu.BsonIdToSId((*v)["_id"])
 			query := map[string]interface{}{
 				"isaudit": true,

src/jy/extract/extractudp.go

@@ -4,8 +4,6 @@ package extract
 import (
 	"encoding/json"
 	"fmt"
-	log "github.com/donnie4w/go-logger/logger"
-	"gopkg.in/mgo.v2/bson"
 	"io/ioutil"
 	"jy/cluster"
 	db "jy/mongodbutil"
@@ -17,6 +15,9 @@ import (
 	qu "qfw/util"
 	"strings"
 	"sync"
+
+	log "github.com/donnie4w/go-logger/logger"
+	"gopkg.in/mgo.v2/bson"
 )
 
 var Udpclient mu.UdpClient //udp对象