README.md 23 KB
{
"udpport": ":19097",
"dupdays": 7,
"mongodb": {
"addr": "172.17.4.85:27080",
"pool": 10,
"db": "qfw",
"extract": "result_20200715",
"extract_back": "result_20200714",
"site": {
"dbname": "qfw",
"coll": "site"
}
},
"task_mongodb": {
"task_addrName": "172.17.4.187:27081",
"task_dbName": "qfw",
"task_collName": "ocr_flie_over",
"pool": 5
},
"jkmail": {
"to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
"api": "http://10.171.112.160:19281/_send/_mail"
},
"nextNode": [
{
"addr": "172.17.4.194",
"port": 1782,
"stype": "project",
"memo": "合并项目"
},
{
"addr": "127.0.0.1",
"port": 1783,
"stype": "bidding",
"memo": "创建招标数据索引new"
}
],
"threads": 1,
"isMerger": false,
"lowHeavy":true,
"timingTask":false,
"timingSpanDay": 5,
"timingPubScope": 720,
"specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
"specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
"specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
"specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
"beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
}
mgo = &MongodbSim{
MongodbAddr: "172.17.4.187:27083",
DbName: "qfw",
Size: 10,
}
mgo.InitPool()
return
func moveTimeoutData() {
log.Println("部署迁移定时任务")
c := cron.New()
c.AddFunc("0 0 0 * * ?", func() { moveOnceTimeOut() })
c.Start()
}
func moveOnceTimeOut() {
log.Println("执行一次迁移超时数据")
sess := mgo.GetMgoConn()
defer mgo.DestoryMongoConn(sess)
now:=time.Now()
move_time := time.Date(now.Year()-2, now.Month(), now.Day(), now.Hour(), 0, 0, 0, time.Local).Unix()
q := map[string]interface{}{
"comeintime": map[string]interface{}{
"$lt": move_time,
},
}
log.Println(q)
it := sess.DB(mgo.DbName).C("result_20200714").Find(&q).Iter()
index := 0
for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
if index%10000 == 0 {
log.Println("index", index)
}
del_id:=BsonTOStringId(tmp["_id"])
mgo.Save("result_20200713", tmp)
mgo.DeleteById("result_20200714",del_id)
tmp = map[string]interface{}{}
}
log.Println("save and delete", " ok index", index)
}
{
"udpport": ":1785",
"dupdays": 7,
"mongodb": {
"addr": "172.17.4.85:27080",
"pool": 10,
"db": "qfw",
"extract": "result_20200715",
"extract_back": "result_20200714",
"site": {
"dbname": "qfw",
"coll": "site"
}
},
"jkmail": {
"to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
"api": "http://172.17.145.179:19281/_send/_mail"
},
"nextNode": [
{
"addr": "127.0.0.1",
"port": 1783,
"stype": "bidding",
"memo": "创建招标数据索引new"
}
],
"threads": 1,
"isMerger": false,
"lowHeavy":true,
"timingTask":false,
"timingSpanDay": 5,
"timingPubScope": 720,
"specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
"specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
"specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
"specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
"beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
}
func historyTaskDay() {
defer util.Catch()
for {
start:=time.Now().Unix()
if gtid=="" {
log.Println("请传gtid,否则无法运行")
os.Exit(0)
return
}
if lteid!="" {
//先进行数据迁移
log.Println("开启一次迁移任务",gtid,lteid)
moveHistoryData(gtid,lteid)
gtid = lteid //替换数据
}
//查询表最后一个id
task_sess := task_mgo.GetMgoConn()
defer task_mgo.DestoryMongoConn(task_sess)
q:=map[string]interface{}{
"isused":true,
}
between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
it_last := task_sess.DB(task_mgo.DbName).C(task_collName).Find(&q).Sort("-_id").Iter()
for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
lteid = util.ObjToString(tmp["gtid"])
log.Println("查询的最后一个任务Id:",lteid)
break
}
log.Println("查询完毕-先睡眠5分钟",gtid,lteid)
time.Sleep(5 * time.Minute)
sess := mgo.GetMgoConn()//连接器
defer mgo.DestoryMongoConn(sess)
//开始判重
q = map[string]interface{}{
"_id": map[string]interface{}{
"$gt": StringTOBsonId(gtid),
"$lte": StringTOBsonId(lteid),
},
}
log.Println("历史判重查询条件：",q,"时间：", between_time)
it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
num,oknum,outnum, deterTime:= int64(0),int64(0),int64(0),int64(0) //计数
updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
pendAllArr:=[][]map[string]interface{}{}//待处理数组
dayArr := []map[string]interface{}{}
for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
if num%10000 == 0 {
log.Println("正序遍历:", num)
}
source := util.ObjToMap(tmp["jsondata"])
if util.IntAll((*source)["sourcewebsite"]) == 1 {
outnum++
updatelock.Lock()
updateExtract = append(updateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat": 1,
"dataging": 0,
"repeat_reason": "sourcewebsite为1 重复",
},
},
})
if len(updateExtract) >= 200 {
log.Println("sourcewebsite,批量更新")
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
tmp = make(map[string]interface{})
continue
}
//取-符合-发布时间X年内的数据
updatelock.Lock()
if util.IntAll(tmp["dataging"]) == 1 {
pubtime := util.Int64All(tmp["publishtime"])
if pubtime > 0 && pubtime >= between_time {
oknum++
if deterTime==0 {
log.Println("找到第一条符合条件的数据")
deterTime = util.Int64All(tmp["publishtime"])
dayArr = append(dayArr,tmp)
}else {
if pubtime-deterTime >timingSpanDay*86400 {
//新数组重新构建，当前组数据加到全部组数据
pendAllArr = append(pendAllArr,dayArr)
dayArr = []map[string]interface{}{}
deterTime = util.Int64All(tmp["publishtime"])
dayArr = append(dayArr,tmp)
}else {
dayArr = append(dayArr,tmp)
}
}
}else {
outnum++
//不在两年内的也清标记
updateExtract = append(updateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"dataging": 0,
},
},
})
if len(updateExtract) >= 200 {
log.Println("不在周期内符合dataging==1,批量更新")
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
}
}
updatelock.Unlock()
tmp = make(map[string]interface{})
}
//批量更新标记
updatelock.Lock()
if len(updateExtract) > 0 {
log.Println("分组后,最后更新不进行判重的数据：",len(updateExtract),oknum+outnum)
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
if len(dayArr)>0 {
pendAllArr = append(pendAllArr,dayArr)
dayArr = []map[string]interface{}{}
}
log.Println("查询数量：",num,"符合条件：",oknum)
if len(pendAllArr) <= 0 {
log.Println("没找到dataging==1的数据")
}
//测试分组数量是否正确
testNum:=0
for k,v:=range pendAllArr {
log.Println("第",k,"组--","数量：",len(v))
testNum = testNum+len(v)
}
log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)
n, repeateN := 0, 0
log.Println("线程数：",threadNum)
pool := make(chan bool, threadNum)
wg := &sync.WaitGroup{}
for k,v:=range pendAllArr { //每组结束更新一波数据
pool <- true
wg.Add(1)
go func(k int, v []map[string]interface{}) {
defer func() {
<-pool
wg.Done()
}()
//每组临时数组 - 互不干扰
groupUpdateExtract := [][]map[string]interface{}{}
//
groupOtherExtract := [][]map[string]interface{}{}
//构建当前组的数据池
log.Println("构建第", k, "组---（数据池）")
//当前组的第一个发布时间
first_pt := util.Int64All(v[len(v)-1]["publishtime"])
curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
log.Println("开始遍历判重第", k, "组 共计数量：", len(v))
n = n + len(v)
log.Println("统计目前总数量：", n, "重复数量：", repeateN)
for _, tmp := range v {
info := NewInfo(tmp)
b, source, reason := curTM.check(info)
if b { //有重复,生成更新语句，更新抽取和更新招标
repeateN++
//重复数据打标签
repeat_ids:=source.repeat_ids
repeat_ids = append(repeat_ids,info.id)
source.repeat_ids = repeat_ids
//替换数据池-更新
DM.replacePoolData(source)
updatelock.Lock()
//更新数据源- 14 或者 15
//判断是否在当前段落
if judgeIsCurIds(gtid,lteid,source.id) {
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{//重复数据打标签
map[string]interface{}{
"_id": StringTOBsonId(source.id),
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat_ids": repeat_ids,
},
},
})
}else {
groupOtherExtract = append(groupOtherExtract, []map[string]interface{}{//重复数据打标签
map[string]interface{}{
"_id": StringTOBsonId(source.id),
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat_ids": repeat_ids,
},
},
})
}
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat": 1,
"repeat_reason": reason,
"repeat_id": source.id,
"dataging": 0,
},
},
})
if len(groupUpdateExtract) >= 500 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
groupUpdateExtract = [][]map[string]interface{}{}
}
if len(groupOtherExtract) >= 500 {
mgo.UpSertBulk(extract_back, groupOtherExtract...)
groupOtherExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
} else {
updatelock.Lock()
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"dataging": 0, //符合条件的都为dataging==0
},
},
})
if len(groupUpdateExtract) >= 500 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
groupUpdateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
}
}
//每组数据结束-更新数据
updatelock.Lock()
if len(groupUpdateExtract) > 0 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
}
if len(groupOtherExtract) > 0 {
mgo.UpSertBulk(extract_back, groupOtherExtract...)
}
updatelock.Unlock()
}(k, v)
}
wg.Wait()
//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
if n >= repeateN && gtid!=lteid{
for _, to := range nextNode {
next_sid := util.BsonIdToSId(gtid)
next_eid := util.BsonIdToSId(lteid)
key := next_sid + "-" + next_eid + "-" + util.ObjToString(to["stype"])
by, _ := json.Marshal(map[string]interface{}{
"gtid": next_sid,
"lteid": next_eid,
"stype": util.ObjToString(to["stype"]),
"key": key,
})
addr := &net.UDPAddr{
IP: net.ParseIP(to["addr"].(string)),
Port: util.IntAll(to["port"]),
}
node := &udpNode{by, addr, time.Now().Unix(), 0}
udptaskmap.Store(key, node)
udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
}
}
end:=time.Now().Unix()
log.Println("this timeTask over.", n, "repeateN:", repeateN,gtid,lteid)
log.Println(gtid,lteid)
if end-start<60*5 {
log.Println("睡眠.............")
time.Sleep(5 * time.Minute)
}
log.Println("继续下一段的历史判重")
}
}func historyTaskDay() {
defer util.Catch()
for {
start:=time.Now().Unix()
if gtid=="" {
log.Println("请传gtid,否则无法运行")
os.Exit(0)
return
}
if lteid!="" {
//先进行数据迁移
log.Println("开启一次迁移任务",gtid,lteid)
moveHistoryData(gtid,lteid)
gtid = lteid //替换数据
}
//查询表最后一个id
task_sess := task_mgo.GetMgoConn()
defer task_mgo.DestoryMongoConn(task_sess)
q:=map[string]interface{}{
"isused":true,
}
between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
it_last := task_sess.DB(task_mgo.DbName).C(task_collName).Find(&q).Sort("-_id").Iter()
for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
lteid = util.ObjToString(tmp["gtid"])
log.Println("查询的最后一个任务Id:",lteid)
break
}
log.Println("查询完毕-先睡眠5分钟",gtid,lteid)
time.Sleep(5 * time.Minute)
sess := mgo.GetMgoConn()//连接器
defer mgo.DestoryMongoConn(sess)
//开始判重
q = map[string]interface{}{
"_id": map[string]interface{}{
"$gt": StringTOBsonId(gtid),
"$lte": StringTOBsonId(lteid),
},
}
log.Println("历史判重查询条件：",q,"时间：", between_time)
it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
num,oknum,outnum, deterTime:= int64(0),int64(0),int64(0),int64(0) //计数
updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
pendAllArr:=[][]map[string]interface{}{}//待处理数组
dayArr := []map[string]interface{}{}
for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
if num%10000 == 0 {
log.Println("正序遍历:", num)
}
source := util.ObjToMap(tmp["jsondata"])
if util.IntAll((*source)["sourcewebsite"]) == 1 {
outnum++
updatelock.Lock()
updateExtract = append(updateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat": 1,
"dataging": 0,
"repeat_reason": "sourcewebsite为1 重复",
},
},
})
if len(updateExtract) >= 200 {
log.Println("sourcewebsite,批量更新")
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
tmp = make(map[string]interface{})
continue
}
//取-符合-发布时间X年内的数据
updatelock.Lock()
if util.IntAll(tmp["dataging"]) == 1 {
pubtime := util.Int64All(tmp["publishtime"])
if pubtime > 0 && pubtime >= between_time {
oknum++
if deterTime==0 {
log.Println("找到第一条符合条件的数据")
deterTime = util.Int64All(tmp["publishtime"])
dayArr = append(dayArr,tmp)
}else {
if pubtime-deterTime >timingSpanDay*86400 {
//新数组重新构建，当前组数据加到全部组数据
pendAllArr = append(pendAllArr,dayArr)
dayArr = []map[string]interface{}{}
deterTime = util.Int64All(tmp["publishtime"])
dayArr = append(dayArr,tmp)
}else {
dayArr = append(dayArr,tmp)
}
}
}else {
outnum++
//不在两年内的也清标记
updateExtract = append(updateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"dataging": 0,
},
},
})
if len(updateExtract) >= 200 {
log.Println("不在周期内符合dataging==1,批量更新")
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
}
}
updatelock.Unlock()
tmp = make(map[string]interface{})
}
//批量更新标记
updatelock.Lock()
if len(updateExtract) > 0 {
log.Println("分组后,最后更新不进行判重的数据：",len(updateExtract),oknum+outnum)
mgo.UpSertBulk(extract, updateExtract...)
updateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
if len(dayArr)>0 {
pendAllArr = append(pendAllArr,dayArr)
dayArr = []map[string]interface{}{}
}
log.Println("查询数量：",num,"符合条件：",oknum)
if len(pendAllArr) <= 0 {
log.Println("没找到dataging==1的数据")
}
//测试分组数量是否正确
testNum:=0
for k,v:=range pendAllArr {
log.Println("第",k,"组--","数量：",len(v))
testNum = testNum+len(v)
}
log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)
n, repeateN := 0, 0
log.Println("线程数：",threadNum)
pool := make(chan bool, threadNum)
wg := &sync.WaitGroup{}
for k,v:=range pendAllArr { //每组结束更新一波数据
pool <- true
wg.Add(1)
go func(k int, v []map[string]interface{}) {
defer func() {
<-pool
wg.Done()
}()
//每组临时数组 - 互不干扰
groupUpdateExtract := [][]map[string]interface{}{}
//
groupOtherExtract := [][]map[string]interface{}{}
//构建当前组的数据池
log.Println("构建第", k, "组---（数据池）")
//当前组的第一个发布时间
first_pt := util.Int64All(v[len(v)-1]["publishtime"])
curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
log.Println("开始遍历判重第", k, "组 共计数量：", len(v))
n = n + len(v)
log.Println("统计目前总数量：", n, "重复数量：", repeateN)
for _, tmp := range v {
info := NewInfo(tmp)
b, source, reason := curTM.check(info)
if b { //有重复,生成更新语句，更新抽取和更新招标
repeateN++
//重复数据打标签
repeat_ids:=source.repeat_ids
repeat_ids = append(repeat_ids,info.id)
source.repeat_ids = repeat_ids
//替换数据池-更新
DM.replacePoolData(source)
updatelock.Lock()
//更新数据源- 14 或者 15
//判断是否在当前段落
if judgeIsCurIds(gtid,lteid,source.id) {
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{//重复数据打标签
map[string]interface{}{
"_id": StringTOBsonId(source.id),
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat_ids": repeat_ids,
},
},
})
}else {
groupOtherExtract = append(groupOtherExtract, []map[string]interface{}{//重复数据打标签
map[string]interface{}{
"_id": StringTOBsonId(source.id),
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat_ids": repeat_ids,
},
},
})
}
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"repeat": 1,
"repeat_reason": reason,
"repeat_id": source.id,
"dataging": 0,
},
},
})
if len(groupUpdateExtract) >= 500 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
groupUpdateExtract = [][]map[string]interface{}{}
}
if len(groupOtherExtract) >= 500 {
mgo.UpSertBulk(extract_back, groupOtherExtract...)
groupOtherExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
} else {
updatelock.Lock()
groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
map[string]interface{}{
"_id": tmp["_id"],
},
map[string]interface{}{
"$set": map[string]interface{}{
"dataging": 0, //符合条件的都为dataging==0
},
},
})
if len(groupUpdateExtract) >= 500 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
groupUpdateExtract = [][]map[string]interface{}{}
}
updatelock.Unlock()
}
}
//每组数据结束-更新数据
updatelock.Lock()
if len(groupUpdateExtract) > 0 {
mgo.UpSertBulk(extract, groupUpdateExtract...)
}
if len(groupOtherExtract) > 0 {
mgo.UpSertBulk(extract_back, groupOtherExtract...)
}
updatelock.Unlock()
}(k, v)
}
wg.Wait()
//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
if n >= repeateN && gtid!=lteid{
for _, to := range nextNode {
next_sid := util.BsonIdToSId(gtid)
next_eid := util.BsonIdToSId(lteid)
key := next_sid + "-" + next_eid + "-" + util.ObjToString(to["stype"])
by, _ := json.Marshal(map[string]interface{}{
"gtid": next_sid,
"lteid": next_eid,
"stype": util.ObjToString(to["stype"]),
"key": key,
})
addr := &net.UDPAddr{
IP: net.ParseIP(to["addr"].(string)),
Port: util.IntAll(to["port"]),
}
node := &udpNode{by, addr, time.Now().Unix(), 0}
udptaskmap.Store(key, node)
udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
}
}
end:=time.Now().Unix()
log.Println("this timeTask over.", n, "repeateN:", repeateN,gtid,lteid)
log.Println(gtid,lteid)
if end-start<60*5 {
log.Println("睡眠.............")
time.Sleep(5 * time.Minute)
}
log.Println("继续下一段的历史判重")
}
}