main.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784
  1. package main
  2. import (
  3. "encoding/json"
  4. "esindex/config"
  5. "esindex/oss"
  6. "fmt"
  7. "github.com/robfig/cron"
  8. "go.uber.org/zap"
  9. "io/ioutil"
  10. util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  11. "jygit.jydev.jianyu360.cn/data_processing/common_utils/elastic"
  12. "jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
  13. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  14. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mysqldb"
  15. "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
  16. "net"
  17. "net/http"
  18. "os"
  19. "os/signal"
  20. "sync"
  21. "syscall"
  22. "time"
  23. )
  24. var (
  25. MgoB *mongodb.MongodbSim
  26. MgoBOld *mongodb.MongodbSim
  27. MgoP *mongodb.MongodbSim
  28. MgoQ *mongodb.MongodbSim
  29. MgoS *mongodb.MongodbSim
  30. Mysql *mysqldb.Mysql
  31. Es, Es2, Es3 *elastic.Elastic //Es3 迁移华为云新集群地址
  32. //PreEs = make(map[string]*elastic.Elastic, 0) //预处理 索引客户端
  33. UdpClient udp.UdpClient
  34. UdpTaskMap = &sync.Map{}
  35. JyUdpAddr *net.UDPAddr
  36. NeUdpAddr *net.UDPAddr
  37. EsBulkSize = 50 // es批量保存大小
  38. updateBiddingPool = make(chan map[string]interface{}, 5000) //更新bingding数据
  39. updateBiddingSp = make(chan bool, 5)
  40. saveEsPool = make(chan map[string]interface{}, 5000) //保存binding数据到es
  41. saveEsSp = make(chan bool, 5)
  42. saveProjectEsPool = make(chan map[string]interface{}, 5000) //保存project数据到es
  43. saveProjectSp = make(chan bool, 5)
  44. saveProjectDetailEsPool = make(chan map[string]interface{}, 5000) //保存project detail 数据到es
  45. saveProjectDetailSp = make(chan bool, 5)
  46. saveEsAllPool = make(chan map[string]interface{}, 5000) //存储单机版es,爬虫采集判重使用
  47. saveEsAllSp = make(chan bool, 5)
  48. saveBiddingAllPool = make(chan map[string]interface{}, 5000) //保存binding数据到es,stype=bidding_all_data
  49. saveBiddingAllBEsSp = make(chan bool, 5)
  50. saveErrBidPool = make(chan map[string]interface{}, 5000)
  51. saveBidSp = make(chan bool, 5)
  52. //detailLength = 50000 // es保存detail长度
  53. fileLength = 200000 // es保存附件文本长度,大于20万 时做一个日志记录
  54. //pscopeLength = 32766 // projectscope长度
  55. specialNames = make(map[string]bool, 0) //存储181 凭安提供的特企,爬虫采购单位
  56. responselock sync.Mutex
  57. BiddingLastNodeResponse int64 //上次节点接受数据时间
  58. ProjectLastNodeResponse int64 //上次节点接受数据时间
  59. )
  60. func init() {
  61. config.InitConf("./common.toml")
  62. InitLog()
  63. InitMysql()
  64. InitMgo()
  65. InitEs()
  66. InitField() //初始化项目数据升索引字段
  67. InitBitmap() //初始化项目名称副标题 bitmap
  68. InitRule() //初始化中国移动定制标签规则
  69. InitEsBiddingField() //初始bidding数据升索引字段
  70. oss.InitOss() // 初始化oss 存储
  71. verifyESFields() //检测es 字段类型
  72. JyUdpAddr = &net.UDPAddr{
  73. IP: net.ParseIP(config.Conf.Udp.JyAddr),
  74. Port: util.IntAll(config.Conf.Udp.JyPort),
  75. }
  76. if config.Conf.Udp.NeAddr != "" {
  77. NeUdpAddr = &net.UDPAddr{
  78. IP: net.ParseIP(config.Conf.Udp.NeAddr),
  79. Port: util.IntAll(config.Conf.Udp.NePort),
  80. }
  81. }
  82. BiddingLastNodeResponse = time.Now().Unix()
  83. ProjectLastNodeResponse = time.Now().Unix()
  84. log.Info("init success")
  85. }
  86. func main() {
  87. //正式环境才执行定时任务
  88. if config.Conf.Env.Stype == 0 {
  89. go LastUdpJob() //监听半小时内有无数据
  90. go checkMapJob() //udp 发送邮件
  91. go task_index() //定时同步更新winner_enterprise、buyer_enterprise ES索引;这个功能很少变动,几乎不需要维护
  92. }
  93. go UpdateBidding() //更新bidding表数据
  94. go SaveEsMethod() //保存es bidding数据
  95. //go SaveBiddingEsMethod() //保存es bidding数据
  96. go SaveAllEsMethod() // 保存爬虫采集临时数据
  97. go SaveProjectEs() //保存项目索引数据
  98. go SaveProjectDetailEs() //保存项目索引,添加了详情字段的新索引
  99. go SaveBiddingAllDataEs() //保存stype=bidding_all_data 数据
  100. go SaveBidErr()
  101. //添加预处理函数
  102. //if config.Conf.Env.OpenPre {
  103. // go SavePreEsMethod()
  104. // go dealPreProcess()
  105. //}
  106. UdpClient = udp.UdpClient{Local: config.Conf.Udp.LocPort, BufSize: 1024}
  107. UdpClient.Listen(processUdpMsg)
  108. log.Info("Udp服务监听", zap.String("port:", config.Conf.Udp.LocPort))
  109. //监听异常退出信号;及时保存项目名称副标题数据
  110. signalChan := make(chan os.Signal, 1)
  111. signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
  112. <-signalChan
  113. saveDb()
  114. //ch := make(chan bool, 1)
  115. //<-ch
  116. }
  117. var pool = make(chan bool, 20)
  118. func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
  119. switch act {
  120. case udp.OP_TYPE_DATA:
  121. var mapInfo map[string]interface{}
  122. err := json.Unmarshal(data, &mapInfo)
  123. log.Info("processUdpMsg", zap.Any("mapInfo:", mapInfo))
  124. if err != nil {
  125. UdpClient.WriteUdp([]byte("err:"+err.Error()), udp.OP_NOOP, ra)
  126. } else if mapInfo != nil {
  127. key, _ := mapInfo["key"].(string)
  128. if key == "" {
  129. key = "udpok"
  130. }
  131. go UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
  132. //有udp 转发时
  133. if config.Conf.Udp.NeAddr != "" {
  134. go SendUdpMsg(mapInfo, NeUdpAddr)
  135. }
  136. tasktype, _ := mapInfo["stype"].(string)
  137. switch tasktype {
  138. case "index-by-id": //单个索引,更新pici
  139. pool <- true
  140. go func() {
  141. defer func() {
  142. <-pool
  143. }()
  144. biddingTaskById(mapInfo)
  145. }()
  146. case "index_by_id": //单个索引,不更新pici
  147. pool <- true
  148. go func() {
  149. defer func() {
  150. <-pool
  151. }()
  152. biddingTaskById(mapInfo)
  153. }()
  154. case "bidding":
  155. BiddingLastNodeResponse = time.Now().Unix()
  156. pool <- true
  157. go func() {
  158. defer func() {
  159. <-pool
  160. }()
  161. biddingTask(mapInfo)
  162. }()
  163. case "biddingall": //补充存量数据
  164. pool <- true
  165. go func() {
  166. defer func() {
  167. <-pool
  168. }()
  169. biddingAllTask(mapInfo)
  170. }()
  171. case "bidding_all_data": //根据biddingall配置文件,存量迁移数据
  172. pool <- true
  173. go func() {
  174. defer func() {
  175. <-pool
  176. }()
  177. biddingAllDataTask()
  178. }()
  179. case "bidding_history":
  180. pool <- true
  181. go func() {
  182. defer func() {
  183. <-pool
  184. }()
  185. biddingTask(mapInfo)
  186. }()
  187. case "project":
  188. ProjectLastNodeResponse = time.Now().Unix()
  189. pool <- true
  190. go func() {
  191. defer func() {
  192. <-pool
  193. }()
  194. projectTask(data, mapInfo)
  195. // 配置项目详情索引
  196. if config.Conf.DB.Es.IndexPD != "" {
  197. projectDetailTask(data, mapInfo)
  198. }
  199. }()
  200. case "project_detail": //添加了详情字段的项目索引
  201. if config.Conf.DB.Es.IndexPD != "" {
  202. ProjectLastNodeResponse = time.Now().Unix()
  203. pool <- true
  204. go func() {
  205. defer func() {
  206. <-pool
  207. }()
  208. projectDetailTask(data, mapInfo)
  209. }()
  210. } else {
  211. log.Info("升级项目索引", zap.String("项目详情索引 ", "缺少项目详情索引配置,请检查配置文件"))
  212. }
  213. case "project_all_data": //存量 projectset 数据
  214. pool <- true
  215. go func() {
  216. defer func() {
  217. <-pool
  218. }()
  219. projectAllData()
  220. }()
  221. case "biddingdata": //es 单机版,采集判重,对应索引 biddingalll; 172.17.4.184:19905
  222. pool <- true
  223. go func() {
  224. defer func() {
  225. <-pool
  226. }()
  227. biddingDataTask(data, mapInfo)
  228. }()
  229. case "biddingdelbyextracttype": //根据bidding表extracttype=-1,删除es中重复数据
  230. pool <- true
  231. go func() {
  232. defer func() {
  233. <-pool
  234. }()
  235. biddingDelByExtracttype(data, mapInfo)
  236. }()
  237. case "buyer_once": // 采购单位昨天增量数据
  238. pool <- true
  239. go func() {
  240. defer func() {
  241. <-pool
  242. }()
  243. buyerOnce()
  244. }()
  245. case "buyer_all": // 采购单位全量数据
  246. pool <- true
  247. go func() {
  248. defer func() {
  249. <-pool
  250. }()
  251. buyerAll()
  252. }()
  253. case "winner_once": // 中标单位昨天增量数据
  254. pool <- true
  255. go func() {
  256. defer func() {
  257. <-pool
  258. }()
  259. winnerEsTaskOnce()
  260. }()
  261. case "winner_all": // 中标单位存量数据
  262. pool <- true
  263. go func() {
  264. defer func() {
  265. <-pool
  266. }()
  267. winnerEsAll()
  268. }()
  269. case "attachment": // 补充附件采集,对应bidding为bidding_downloadfile_log
  270. pool <- true
  271. go func() {
  272. defer func() {
  273. <-pool
  274. }()
  275. //有单独配置其他操作
  276. if len(config.Conf.Others) > 0 {
  277. if v, ok := config.Conf.Others[tasktype]; ok {
  278. attachmentBiddingTask(mapInfo, v)
  279. }
  280. }
  281. }()
  282. default:
  283. pool <- true
  284. go func() {
  285. defer func() {
  286. <-pool
  287. }()
  288. log.Info("err", zap.Any("mapInfo", mapInfo))
  289. }()
  290. }
  291. }
  292. case udp.OP_NOOP: //下个节点回应
  293. ok := string(data)
  294. if ok != "" {
  295. log.Info("udp re", zap.String("data:", ok))
  296. UdpTaskMap.Delete(ok)
  297. }
  298. }
  299. }
  300. func task_index() {
  301. c := cron.New()
  302. _ = c.AddFunc("0 00 00 * * *", func() { task_winneres() }) //每天凌晨执行一次winner生索引
  303. _ = c.AddFunc("0 00 01 * * *", func() { task_buyeres() }) //每天1点执行一次buyer生索引
  304. c.Start()
  305. }
  306. func task_winneres() {
  307. log.Info("定时任务,winneres")
  308. winnerEsTaskOnce()
  309. }
  310. func task_buyeres() {
  311. log.Info("定时任务,buyeres")
  312. buyerOnce()
  313. }
  314. type UdpNode struct {
  315. data []byte
  316. addr *net.UDPAddr
  317. timestamp int64
  318. retry int
  319. }
  320. // UpdateBidding 更新bidding表数据
  321. func UpdateBidding() {
  322. //arru := make([][]map[string]interface{}, 10)
  323. //indexu := 0
  324. for {
  325. select {
  326. case v := <-updateBiddingPool:
  327. MgoB.UpdateById(config.Conf.DB.MongoB.Coll, v["_id"], map[string]interface{}{"$set": v["set"]})
  328. //arru[indexu] = v
  329. //indexu++
  330. //if indexu == 10 {
  331. // updateBiddingSp <- true
  332. //go func(arru [][]map[string]interface{}) {
  333. // defer func() {
  334. // <-updateBiddingSp
  335. //}()
  336. //MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
  337. //}(arru)
  338. //arru = make([][]map[string]interface{}, 10)
  339. //indexu = 0
  340. //}
  341. //case <-time.After(1000 * time.Millisecond):
  342. // if indexu > 0 {
  343. // updateBiddingSp <- true
  344. // go func(arru [][]map[string]interface{}) {
  345. // defer func() {
  346. // <-updateBiddingSp
  347. // }()
  348. // MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
  349. // }(arru[:indexu])
  350. // arru = make([][]map[string]interface{}, 200)
  351. // indexu = 0
  352. // }
  353. }
  354. }
  355. }
  356. // SaveBidErr 记录错误信息,暂时记录 附件过长的
  357. func SaveBidErr() {
  358. arru := make([]map[string]interface{}, 200)
  359. indexu := 0
  360. for {
  361. select {
  362. case v := <-saveErrBidPool:
  363. arru[indexu] = v
  364. indexu++
  365. if indexu == 200 {
  366. saveBidSp <- true
  367. go func(arru []map[string]interface{}) {
  368. defer func() {
  369. <-saveBidSp
  370. }()
  371. MgoB.SaveBulk("bidding_es_err_record", arru...)
  372. }(arru)
  373. arru = make([]map[string]interface{}, 200)
  374. indexu = 0
  375. }
  376. case <-time.After(1000 * time.Millisecond):
  377. if indexu > 0 {
  378. saveBidSp <- true
  379. go func(arru []map[string]interface{}) {
  380. defer func() {
  381. <-saveBidSp
  382. }()
  383. MgoB.SaveBulk("bidding_es_err_record", arru...)
  384. }(arru[:indexu])
  385. arru = make([]map[string]interface{}, 200)
  386. indexu = 0
  387. }
  388. }
  389. }
  390. }
  391. // SaveEsMethod 保存bidding数据到es;单携程保存
  392. func SaveEsMethod() {
  393. for {
  394. select {
  395. case v := <-saveEsPool:
  396. id := v["id"]
  397. ids := v["_id"]
  398. //如果优选,存在上次的优选ID,需要删除上次的ID索引
  399. oid := util.ObjToString(v["old_preferid"])
  400. delete(v, "old_preferid")
  401. if oid != "" {
  402. err := Es.DeleteByID(config.Conf.DB.Es.IndexB, oid)
  403. if err != nil {
  404. log.Info("SaveEsMethod", zap.String("上次优选ID删除失败", oid))
  405. }
  406. }
  407. Es.Save(config.Conf.DB.Es.IndexB, v)
  408. // 华为云集群1
  409. if config.Conf.DB.Es.Addr2 != "" {
  410. v["id"] = id
  411. v["_id"] = ids
  412. delete(v, "old_preferid")
  413. if oid != "" {
  414. err := Es2.DeleteByID(config.Conf.DB.Es.IndexB, oid)
  415. if err != nil {
  416. log.Info("SaveEsMethod", zap.String("上次优选ID删除失败", oid))
  417. }
  418. }
  419. Es2.Save(config.Conf.DB.Es.Indexb2, v)
  420. }
  421. // 华为云新集群2,迁移原来阿里云数据
  422. if config.Conf.DB.Es.Addr3 != "" {
  423. v["id"] = id
  424. v["_id"] = ids
  425. delete(v, "old_preferid")
  426. if oid != "" {
  427. err := Es3.DeleteByID(config.Conf.DB.Es.IndexB, oid)
  428. if err != nil {
  429. log.Info("SaveEsMethod", zap.String("上次优选ID删除失败", oid))
  430. }
  431. }
  432. Es3.Save(config.Conf.DB.Es.Indexb3, v)
  433. }
  434. }
  435. }
  436. }
  437. // SaveBiddingEsMethod 批量保存bidding数据
  438. func SaveBiddingEsMethod() {
  439. arru := make([]map[string]interface{}, EsBulkSize)
  440. indexu := 0
  441. for {
  442. select {
  443. case v := <-saveEsPool:
  444. arru[indexu] = v
  445. indexu++
  446. if indexu == EsBulkSize {
  447. saveEsSp <- true
  448. go func(arru []map[string]interface{}) {
  449. defer func() {
  450. <-saveEsSp
  451. }()
  452. if config.Conf.DB.Es.Addr != "" {
  453. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  454. }
  455. // 集群地址2
  456. if config.Conf.DB.Es.Addr2 != "" {
  457. Es2.BulkSave(config.Conf.DB.Es.Indexb2, arru)
  458. }
  459. // 集群地址3
  460. if config.Conf.DB.Es.Addr3 != "" {
  461. Es3.BulkSave(config.Conf.DB.Es.Indexb3, arru)
  462. }
  463. }(arru)
  464. arru = make([]map[string]interface{}, EsBulkSize)
  465. indexu = 0
  466. }
  467. case <-time.After(1000 * time.Millisecond):
  468. if indexu > 0 {
  469. saveEsSp <- true
  470. go func(arru []map[string]interface{}) {
  471. defer func() {
  472. <-saveEsSp
  473. }()
  474. if config.Conf.DB.Es.Addr != "" {
  475. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  476. }
  477. // 集群地址2
  478. if config.Conf.DB.Es.Addr2 != "" {
  479. Es2.BulkSave(config.Conf.DB.Es.Indexb2, arru)
  480. }
  481. // 集群地址3
  482. if config.Conf.DB.Es.Addr3 != "" {
  483. Es3.BulkSave(config.Conf.DB.Es.Indexb3, arru)
  484. }
  485. }(arru[:indexu])
  486. arru = make([]map[string]interface{}, EsBulkSize)
  487. indexu = 0
  488. }
  489. }
  490. }
  491. }
  492. // SaveAllEsMethod 保存爬虫采集临时数据,保存在华为云上
  493. func SaveAllEsMethod() {
  494. arru := make([]map[string]interface{}, EsBulkSize)
  495. indexu := 0
  496. for {
  497. select {
  498. case v := <-saveEsAllPool:
  499. arru[indexu] = v
  500. indexu++
  501. if indexu == EsBulkSize {
  502. saveEsAllSp <- true
  503. go func(arru []map[string]interface{}) {
  504. defer func() {
  505. <-saveEsAllSp
  506. }()
  507. if config.Conf.DB.Es.Addr2 != "" {
  508. Es2.BulkSave("biddingall", arru)
  509. }
  510. }(arru)
  511. arru = make([]map[string]interface{}, EsBulkSize)
  512. indexu = 0
  513. }
  514. case <-time.After(1000 * time.Millisecond):
  515. if indexu > 0 {
  516. saveEsAllSp <- true
  517. go func(arru []map[string]interface{}) {
  518. defer func() {
  519. <-saveEsAllSp
  520. }()
  521. if config.Conf.DB.Es.Addr2 != "" {
  522. Es2.BulkSave("biddingall", arru)
  523. }
  524. }(arru[:indexu])
  525. arru = make([]map[string]interface{}, EsBulkSize)
  526. indexu = 0
  527. }
  528. }
  529. }
  530. }
  531. // SaveBiddingAllDataEs 保存bidding数据到Es,stype=bidding_all_data 数据
  532. func SaveBiddingAllDataEs() {
  533. arru := make([]map[string]interface{}, EsBulkSize)
  534. indexu := 0
  535. for {
  536. select {
  537. case v := <-saveBiddingAllPool:
  538. arru[indexu] = v
  539. indexu++
  540. if indexu == EsBulkSize {
  541. saveBiddingAllBEsSp <- true
  542. go func(arru []map[string]interface{}) {
  543. defer func() {
  544. <-saveBiddingAllBEsSp
  545. }()
  546. //1.阿里云集群
  547. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  548. }(arru)
  549. arru = make([]map[string]interface{}, EsBulkSize)
  550. indexu = 0
  551. }
  552. case <-time.After(1000 * time.Millisecond):
  553. if indexu > 0 {
  554. saveBiddingAllBEsSp <- true
  555. go func(arru []map[string]interface{}) {
  556. defer func() {
  557. <-saveBiddingAllBEsSp
  558. }()
  559. //1.阿里云集群
  560. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  561. }(arru[:indexu])
  562. arru = make([]map[string]interface{}, EsBulkSize)
  563. indexu = 0
  564. }
  565. }
  566. }
  567. }
  568. // SaveProjectEsMethod 保存项目索引数据
  569. func SaveProjectEsMethod() {
  570. arru := make([]map[string]interface{}, EsBulkSize)
  571. indexu := 0
  572. for {
  573. select {
  574. case v := <-saveProjectEsPool:
  575. arru[indexu] = v
  576. indexu++
  577. if indexu == EsBulkSize {
  578. saveProjectSp <- true
  579. go func(arru []map[string]interface{}) {
  580. defer func() {
  581. <-saveProjectSp
  582. }()
  583. Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
  584. // 华为云新集群,存储标讯、项目、凭安数据
  585. if config.Conf.DB.Es.Addr3 != "" {
  586. Es3.BulkSave(config.Conf.DB.Es.IndexP, arru)
  587. }
  588. }(arru)
  589. arru = make([]map[string]interface{}, EsBulkSize)
  590. indexu = 0
  591. }
  592. case <-time.After(1000 * time.Millisecond):
  593. if indexu > 0 {
  594. saveProjectSp <- true
  595. go func(arru []map[string]interface{}) {
  596. defer func() {
  597. <-saveProjectSp
  598. }()
  599. Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
  600. // 华为云新集群,存储标讯、项目、凭安数据
  601. if config.Conf.DB.Es.Addr3 != "" {
  602. Es3.BulkSave(config.Conf.DB.Es.IndexP, arru)
  603. }
  604. }(arru[:indexu])
  605. arru = make([]map[string]interface{}, EsBulkSize)
  606. indexu = 0
  607. }
  608. }
  609. }
  610. }
  611. // SaveProjectEs 保存项目索引数据,但携程保存
  612. func SaveProjectEs() {
  613. for {
  614. select {
  615. case v := <-saveProjectEsPool:
  616. id := v["id"]
  617. ids := v["_id"]
  618. Es.Save(config.Conf.DB.Es.IndexP, v)
  619. // 华为云新集群,存储标讯、项目、凭安数据
  620. if config.Conf.DB.Es.Addr3 != "" {
  621. v["id"] = id
  622. v["_id"] = ids
  623. Es3.Save(config.Conf.DB.Es.IndexP, v)
  624. }
  625. }
  626. }
  627. }
  628. // SaveProjectDetailEs 保存项目索引,支持详情字段
  629. func SaveProjectDetailEs() {
  630. for {
  631. select {
  632. case v := <-saveProjectDetailEsPool:
  633. id := v["id"]
  634. ids := v["_id"]
  635. Es.Save(config.Conf.DB.Es.IndexPD, v)
  636. // 华为云新集群,存储标讯、项目、凭安数据
  637. if config.Conf.DB.Es.Addr3 != "" {
  638. v["id"] = id
  639. v["_id"] = ids
  640. Es3.Save(config.Conf.DB.Es.IndexPD, v)
  641. }
  642. }
  643. }
  644. }
  645. func checkMapJob() {
  646. if config.Conf.Mail.Send {
  647. log.Info("checkMapJob", zap.String("to:", config.Conf.Mail.To))
  648. for {
  649. UdpTaskMap.Range(func(k, v interface{}) bool {
  650. now := time.Now().Unix()
  651. node, _ := v.(*UdpNode)
  652. if now-node.timestamp > 120 {
  653. node.retry++
  654. if node.retry > 5 {
  655. UdpTaskMap.Delete(k)
  656. res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", config.Conf.Mail.Api, config.Conf.Mail.To, "field-sync-send-fail", k.(string)))
  657. if err == nil {
  658. defer res.Body.Close()
  659. read, err := ioutil.ReadAll(res.Body)
  660. log.Info("send mail ...", zap.String("r:", string(read)), zap.Any("err:", err))
  661. }
  662. } else {
  663. log.Info("udp重发", zap.Any("k:", k))
  664. UdpClient.WriteUdp(node.data, udp.OP_TYPE_DATA, node.addr)
  665. }
  666. } else if now-node.timestamp > 10 {
  667. log.Info("udp任务超时中..", zap.Any("k:", k))
  668. }
  669. return true
  670. })
  671. time.Sleep(60 * time.Second)
  672. }
  673. }
  674. }
  675. func task() {
  676. sess := MgoB.GetMgoConn()
  677. defer MgoB.DestoryMongoConn(sess)
  678. ch := make(chan bool, 10)
  679. wg := &sync.WaitGroup{}
  680. query := sess.DB("qfw").C("result_replace_repair_log").Find(nil).Iter()
  681. count := 0
  682. for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
  683. if count%1000 == 0 {
  684. util.Debug("current ---", count)
  685. }
  686. ch <- true
  687. wg.Add(1)
  688. go func(tmp map[string]interface{}) {
  689. defer func() {
  690. <-ch
  691. wg.Done()
  692. }()
  693. if id := util.ObjToString(tmp["replace_id"]); mongodb.IsObjectIdHex(id) {
  694. biddingTaskById(map[string]interface{}{"infoid": id, "stype": "bidding"})
  695. }
  696. }(tmp)
  697. tmp = make(map[string]interface{})
  698. }
  699. wg.Wait()
  700. util.Debug("over ---", count)
  701. }
  702. // LastUdpJob 处理UDP 没有接受数据
  703. func LastUdpJob() {
  704. for {
  705. responselock.Lock()
  706. if time.Now().Unix()-BiddingLastNodeResponse >= 1800 {
  707. BiddingLastNodeResponse = time.Now().Unix() //重置时间
  708. sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无bidding据进入 ...相关人员检查..."))
  709. }
  710. if time.Now().Unix()-ProjectLastNodeResponse >= 1800 {
  711. ProjectLastNodeResponse = time.Now().Unix() //重置时间
  712. sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无project数据进入 ...相关人员检查..."))
  713. }
  714. responselock.Unlock()
  715. time.Sleep(300 * time.Second)
  716. }
  717. }
  718. // sendErrMailApi 发送邮件
  719. func sendErrMailApi(title, body string) {
  720. var tomail, api string
  721. if config.Conf.Mail.Send {
  722. tomail = config.Conf.Mail.To
  723. api = config.Conf.Mail.Api
  724. }
  725. log.Info("sendErrMailApi", zap.Any(tomail, api))
  726. res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, tomail, title, body))
  727. if err == nil {
  728. defer res.Body.Close()
  729. read, err := ioutil.ReadAll(res.Body)
  730. if err != nil {
  731. log.Info("邮件发送成功", zap.String("read", string(read)))
  732. }
  733. } else {
  734. log.Info("sendErrMailApi", zap.String("邮件发送失败", err.Error()))
  735. }
  736. }