main.go 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590
  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. "sync"
  19. "time"
  20. )
  21. var (
  22. MgoB *mongodb.MongodbSim
  23. MgoP *mongodb.MongodbSim
  24. MgoQ *mongodb.MongodbSim
  25. MgoS *mongodb.MongodbSim
  26. Mysql *mysqldb.Mysql
  27. Es, Es1, Es2 *elastic.Elastic
  28. UdpClient udp.UdpClient
  29. UdpTaskMap = &sync.Map{}
  30. JyUdpAddr *net.UDPAddr
  31. NeUdpAddr *net.UDPAddr
  32. EsBulkSize = 50 // es批量保存大小
  33. updateBiddingPool = make(chan []map[string]interface{}, 5000) //更新bingding数据
  34. updateBiddingSp = make(chan bool, 5)
  35. saveEsPool = make(chan map[string]interface{}, 5000) //保存binding数据到es
  36. saveEsSp = make(chan bool, 5)
  37. saveProjectEsPool = make(chan map[string]interface{}, 5000) //保存project数据到es
  38. saveProjectSp = make(chan bool, 5)
  39. saveEsAllPool = make(chan map[string]interface{}, 5000) //存储单机版es,爬虫采集判重使用
  40. saveEsAllSp = make(chan bool, 5)
  41. saveErrBidPool = make(chan map[string]interface{}, 5000)
  42. saveBidSp = make(chan bool, 5)
  43. //detailLength = 50000 // es保存detail长度
  44. fileLength = 200000 // es保存附件文本长度,大于20万 时做一个日志记录
  45. //pscopeLength = 32766 // projectscope长度
  46. specialNames = make(map[string]bool, 0) //存储181 凭安提供的特企,爬虫采购单位
  47. responselock sync.Mutex
  48. BiddingLastNodeResponse int64 //上次节点接受数据时间
  49. ProjectLastNodeResponse int64 //上次节点接受数据时间
  50. )
  51. func init() {
  52. config.InitConf("./common.toml")
  53. InitLog()
  54. InitMysql()
  55. InitMgo()
  56. InitEs()
  57. InitField()
  58. InitEsBiddingField()
  59. oss.InitOss()
  60. //verifyESFields() //检测es 字段类型
  61. JyUdpAddr = &net.UDPAddr{
  62. IP: net.ParseIP(config.Conf.Udp.JyAddr),
  63. Port: util.IntAll(config.Conf.Udp.JyPort),
  64. }
  65. if config.Conf.Udp.NeAddr != "" {
  66. NeUdpAddr = &net.UDPAddr{
  67. IP: net.ParseIP(config.Conf.Udp.NeAddr),
  68. Port: util.IntAll(config.Conf.Udp.NePort),
  69. }
  70. }
  71. BiddingLastNodeResponse = time.Now().Unix()
  72. ProjectLastNodeResponse = time.Now().Unix()
  73. log.Info("init success")
  74. }
  75. func main() {
  76. //正式环境才执行定时任务
  77. if config.Conf.Env.Stype == 0 {
  78. go LastUdpJob() //监听半小时内有无数据
  79. go checkMapJob() //udp 发送邮件
  80. go task_index() //定时同步更新winner_enterprise、buyer_enterprise ES索引;这个功能很少变动,几乎不需要维护
  81. }
  82. go UpdateBidding() //更新bidding表数据
  83. go SaveEsMethod()
  84. go SaveAllEsMethod()
  85. go SaveProjectEs()
  86. go SaveBidErr()
  87. UdpClient = udp.UdpClient{Local: config.Conf.Udp.LocPort, BufSize: 1024}
  88. UdpClient.Listen(processUdpMsg)
  89. log.Info("Udp服务监听", zap.String("port:", config.Conf.Udp.LocPort))
  90. ch := make(chan bool, 1)
  91. <-ch
  92. }
  93. var pool = make(chan bool, 20)
  94. func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
  95. switch act {
  96. case udp.OP_TYPE_DATA:
  97. var mapInfo map[string]interface{}
  98. err := json.Unmarshal(data, &mapInfo)
  99. log.Info("processUdpMsg", zap.Any("mapInfo:", mapInfo))
  100. if err != nil {
  101. UdpClient.WriteUdp([]byte("err:"+err.Error()), udp.OP_NOOP, ra)
  102. } else if mapInfo != nil {
  103. key, _ := mapInfo["key"].(string)
  104. if key == "" {
  105. key = "udpok"
  106. }
  107. go UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
  108. //有udp 转发时
  109. if config.Conf.Udp.NeAddr != "" {
  110. go SendUdpMsg(mapInfo, NeUdpAddr)
  111. }
  112. tasktype, _ := mapInfo["stype"].(string)
  113. switch tasktype {
  114. case "index-by-id": //单个索引,更新pici
  115. pool <- true
  116. go func() {
  117. defer func() {
  118. <-pool
  119. }()
  120. biddingTaskById(mapInfo)
  121. }()
  122. case "index_by_id": //单个索引,不更新pici
  123. pool <- true
  124. go func() {
  125. defer func() {
  126. <-pool
  127. }()
  128. biddingTaskById(mapInfo)
  129. }()
  130. case "bidding":
  131. BiddingLastNodeResponse = time.Now().Unix()
  132. pool <- true
  133. go func() {
  134. defer func() {
  135. <-pool
  136. }()
  137. biddingTask(mapInfo)
  138. }()
  139. case "biddingall": //补充存量数据
  140. pool <- true
  141. go func() {
  142. defer func() {
  143. <-pool
  144. }()
  145. biddingAllTask(mapInfo)
  146. }()
  147. case "bidding_all_data": //根据biddingall配置文件,存量迁移数据
  148. pool <- true
  149. go func() {
  150. defer func() {
  151. <-pool
  152. }()
  153. biddingAllDataTask()
  154. }()
  155. case "bidding_history":
  156. pool <- true
  157. go func() {
  158. defer func() {
  159. <-pool
  160. }()
  161. biddingTask(mapInfo)
  162. }()
  163. case "project":
  164. ProjectLastNodeResponse = time.Now().Unix()
  165. pool <- true
  166. go func() {
  167. defer func() {
  168. <-pool
  169. }()
  170. projectTask(data, mapInfo)
  171. }()
  172. case "project_all_data": //存量 projectset 数据
  173. pool <- true
  174. go func() {
  175. defer func() {
  176. <-pool
  177. }()
  178. projectAllData()
  179. }()
  180. case "biddingdata": //es 单机版,采集判重
  181. pool <- true
  182. go func() {
  183. defer func() {
  184. <-pool
  185. }()
  186. biddingDataTask(data, mapInfo)
  187. }()
  188. case "biddingdelbyextracttype": //根据bidding表extracttype=-1,删除es中重复数据
  189. pool <- true
  190. go func() {
  191. defer func() {
  192. <-pool
  193. }()
  194. biddingDelByExtracttype(data, mapInfo)
  195. }()
  196. case "buyer_once": // 采购单位昨天增量数据
  197. pool <- true
  198. go func() {
  199. defer func() {
  200. <-pool
  201. }()
  202. buyerOnce()
  203. }()
  204. case "winner_once": // 中标单位昨天增量数据
  205. pool <- true
  206. go func() {
  207. defer func() {
  208. <-pool
  209. }()
  210. winnerEsTaskOnce()
  211. }()
  212. case "winner_all": // 中标单位存量数据
  213. pool <- true
  214. go func() {
  215. defer func() {
  216. <-pool
  217. }()
  218. winnerEsAll()
  219. }()
  220. case "attachment": // 补充附件采集,对应bidding为bidding_downloadfile_log
  221. pool <- true
  222. go func() {
  223. defer func() {
  224. <-pool
  225. }()
  226. //有单独配置其他操作
  227. if len(config.Conf.Others) > 0 {
  228. if v, ok := config.Conf.Others[tasktype]; ok {
  229. attachmentBiddingTask(mapInfo, v)
  230. }
  231. }
  232. }()
  233. default:
  234. pool <- true
  235. go func() {
  236. defer func() {
  237. <-pool
  238. }()
  239. log.Info("err", zap.Any("mapInfo", mapInfo))
  240. }()
  241. }
  242. }
  243. case udp.OP_NOOP: //下个节点回应
  244. ok := string(data)
  245. if ok != "" {
  246. log.Info("udp re", zap.String("data:", ok))
  247. UdpTaskMap.Delete(ok)
  248. }
  249. }
  250. }
  251. func task_index() {
  252. c := cron.New()
  253. _ = c.AddFunc("0 0 0 * * ?", func() { task_winneres() }) //每天凌晨执行一次winner生索引
  254. _ = c.AddFunc("0 0 1 * * ?", func() { task_buyeres() }) //每天1点执行一次buyer生索引
  255. c.Start()
  256. }
  257. func task_winneres() {
  258. log.Info("定时任务,winneres")
  259. winnerEsTaskOnce()
  260. }
  261. func task_buyeres() {
  262. log.Info("定时任务,buyeres")
  263. buyerOnce()
  264. }
  265. type UdpNode struct {
  266. data []byte
  267. addr *net.UDPAddr
  268. timestamp int64
  269. retry int
  270. }
  271. //UpdateBidding 更新bidding表数据
  272. func UpdateBidding() {
  273. arru := make([][]map[string]interface{}, 200)
  274. indexu := 0
  275. for {
  276. select {
  277. case v := <-updateBiddingPool:
  278. arru[indexu] = v
  279. indexu++
  280. if indexu == 200 {
  281. updateBiddingSp <- true
  282. go func(arru [][]map[string]interface{}) {
  283. defer func() {
  284. <-updateBiddingSp
  285. }()
  286. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
  287. }(arru)
  288. arru = make([][]map[string]interface{}, 200)
  289. indexu = 0
  290. }
  291. case <-time.After(1000 * time.Millisecond):
  292. if indexu > 0 {
  293. updateBiddingSp <- true
  294. go func(arru [][]map[string]interface{}) {
  295. defer func() {
  296. <-updateBiddingSp
  297. }()
  298. MgoB.UpdateBulk(config.Conf.DB.MongoB.Coll, arru...)
  299. }(arru[:indexu])
  300. arru = make([][]map[string]interface{}, 200)
  301. indexu = 0
  302. }
  303. }
  304. }
  305. }
  306. //SaveBidErr 记录错误信息,暂时记录 附件过长的
  307. func SaveBidErr() {
  308. arru := make([]map[string]interface{}, 200)
  309. indexu := 0
  310. for {
  311. select {
  312. case v := <-saveErrBidPool:
  313. arru[indexu] = v
  314. indexu++
  315. if indexu == 200 {
  316. saveBidSp <- true
  317. go func(arru []map[string]interface{}) {
  318. defer func() {
  319. <-saveBidSp
  320. }()
  321. MgoB.SaveBulk("bidding_es_err_record", arru...)
  322. }(arru)
  323. arru = make([]map[string]interface{}, 200)
  324. indexu = 0
  325. }
  326. case <-time.After(1000 * time.Millisecond):
  327. if indexu > 0 {
  328. saveBidSp <- true
  329. go func(arru []map[string]interface{}) {
  330. defer func() {
  331. <-saveBidSp
  332. }()
  333. MgoB.SaveBulk("bidding_es_err_record", arru...)
  334. }(arru[:indexu])
  335. arru = make([]map[string]interface{}, 200)
  336. indexu = 0
  337. }
  338. }
  339. }
  340. }
  341. //SaveEsMethod 保存到es
  342. func SaveEsMethod() {
  343. arru := make([]map[string]interface{}, EsBulkSize)
  344. indexu := 0
  345. for {
  346. select {
  347. case v := <-saveEsPool:
  348. arru[indexu] = v
  349. indexu++
  350. if indexu == EsBulkSize {
  351. saveEsSp <- true
  352. go func(arru []map[string]interface{}) {
  353. defer func() {
  354. <-saveEsSp
  355. }()
  356. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  357. if config.Conf.DB.Es.IndexTmp != "" {
  358. Es2.BulkSave(config.Conf.DB.Es.IndexTmp, arru) // 新集群
  359. Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru) //老集群
  360. }
  361. if config.Conf.DB.Es.Addr2 != "" {
  362. Es2.BulkSave(config.Conf.DB.Es.Indexb2, arru)
  363. }
  364. }(arru)
  365. arru = make([]map[string]interface{}, EsBulkSize)
  366. indexu = 0
  367. }
  368. case <-time.After(1000 * time.Millisecond):
  369. if indexu > 0 {
  370. saveEsSp <- true
  371. go func(arru []map[string]interface{}) {
  372. defer func() {
  373. <-saveEsSp
  374. }()
  375. Es.BulkSave(config.Conf.DB.Es.IndexB, arru)
  376. if config.Conf.DB.Es.IndexTmp != "" {
  377. Es2.BulkSave(config.Conf.DB.Es.IndexTmp, arru)
  378. Es.BulkSave(config.Conf.DB.Es.IndexTmp, arru)
  379. }
  380. if config.Conf.DB.Es.Addr2 != "" {
  381. Es2.BulkSave(config.Conf.DB.Es.Indexb2, arru)
  382. }
  383. }(arru[:indexu])
  384. arru = make([]map[string]interface{}, EsBulkSize)
  385. indexu = 0
  386. }
  387. }
  388. }
  389. }
  390. func SaveAllEsMethod() {
  391. arru := make([]map[string]interface{}, EsBulkSize)
  392. indexu := 0
  393. for {
  394. select {
  395. case v := <-saveEsAllPool:
  396. arru[indexu] = v
  397. indexu++
  398. if indexu == EsBulkSize {
  399. saveEsAllSp <- true
  400. go func(arru []map[string]interface{}) {
  401. defer func() {
  402. <-saveEsAllSp
  403. }()
  404. Es1.BulkSave("biddingall", arru)
  405. }(arru)
  406. arru = make([]map[string]interface{}, EsBulkSize)
  407. indexu = 0
  408. }
  409. case <-time.After(1000 * time.Millisecond):
  410. if indexu > 0 {
  411. saveEsAllSp <- true
  412. go func(arru []map[string]interface{}) {
  413. defer func() {
  414. <-saveEsAllSp
  415. }()
  416. Es1.BulkSave("biddingall", arru)
  417. }(arru[:indexu])
  418. arru = make([]map[string]interface{}, EsBulkSize)
  419. indexu = 0
  420. }
  421. }
  422. }
  423. }
  424. func SaveProjectEs() {
  425. arru := make([]map[string]interface{}, EsBulkSize)
  426. indexu := 0
  427. for {
  428. select {
  429. case v := <-saveProjectEsPool:
  430. arru[indexu] = v
  431. indexu++
  432. if indexu == EsBulkSize {
  433. saveProjectSp <- true
  434. go func(arru []map[string]interface{}) {
  435. defer func() {
  436. <-saveProjectSp
  437. }()
  438. Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
  439. }(arru)
  440. arru = make([]map[string]interface{}, EsBulkSize)
  441. indexu = 0
  442. }
  443. case <-time.After(1000 * time.Millisecond):
  444. if indexu > 0 {
  445. saveProjectSp <- true
  446. go func(arru []map[string]interface{}) {
  447. defer func() {
  448. <-saveProjectSp
  449. }()
  450. Es.BulkSave(config.Conf.DB.Es.IndexP, arru)
  451. }(arru[:indexu])
  452. arru = make([]map[string]interface{}, EsBulkSize)
  453. indexu = 0
  454. }
  455. }
  456. }
  457. }
  458. func checkMapJob() {
  459. if config.Conf.Mail.Send {
  460. log.Info("checkMapJob", zap.String("to:", config.Conf.Mail.To))
  461. for {
  462. UdpTaskMap.Range(func(k, v interface{}) bool {
  463. now := time.Now().Unix()
  464. node, _ := v.(*UdpNode)
  465. if now-node.timestamp > 120 {
  466. node.retry++
  467. if node.retry > 5 {
  468. UdpTaskMap.Delete(k)
  469. 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)))
  470. if err == nil {
  471. defer res.Body.Close()
  472. read, err := ioutil.ReadAll(res.Body)
  473. log.Info("send mail ...", zap.String("r:", string(read)), zap.Any("err:", err))
  474. }
  475. } else {
  476. log.Info("udp重发", zap.Any("k:", k))
  477. UdpClient.WriteUdp(node.data, udp.OP_TYPE_DATA, node.addr)
  478. }
  479. } else if now-node.timestamp > 10 {
  480. log.Info("udp任务超时中..", zap.Any("k:", k))
  481. }
  482. return true
  483. })
  484. time.Sleep(60 * time.Second)
  485. }
  486. }
  487. }
  488. func task() {
  489. sess := MgoB.GetMgoConn()
  490. defer MgoB.DestoryMongoConn(sess)
  491. ch := make(chan bool, 10)
  492. wg := &sync.WaitGroup{}
  493. query := sess.DB("qfw").C("result_replace_repair_log").Find(nil).Iter()
  494. count := 0
  495. for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
  496. if count%1000 == 0 {
  497. util.Debug("current ---", count)
  498. }
  499. ch <- true
  500. wg.Add(1)
  501. go func(tmp map[string]interface{}) {
  502. defer func() {
  503. <-ch
  504. wg.Done()
  505. }()
  506. if id := util.ObjToString(tmp["replace_id"]); mongodb.IsObjectIdHex(id) {
  507. biddingTaskById(map[string]interface{}{"infoid": id, "stype": "bidding"})
  508. }
  509. }(tmp)
  510. tmp = make(map[string]interface{})
  511. }
  512. wg.Wait()
  513. util.Debug("over ---", count)
  514. }
  515. //LastUdpJob 处理UDP 没有接受数据
  516. func LastUdpJob() {
  517. for {
  518. responselock.Lock()
  519. if time.Now().Unix()-BiddingLastNodeResponse >= 1800 {
  520. BiddingLastNodeResponse = time.Now().Unix() //重置时间
  521. sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无bidding据进入 ...相关人员检查..."))
  522. }
  523. if time.Now().Unix()-ProjectLastNodeResponse >= 1800 {
  524. ProjectLastNodeResponse = time.Now().Unix() //重置时间
  525. sendErrMailApi("索引程序异常", fmt.Sprintf("半小时左右 无project数据进入 ...相关人员检查..."))
  526. }
  527. responselock.Unlock()
  528. time.Sleep(300 * time.Second)
  529. }
  530. }
  531. //sendErrMailApi 发送邮件
  532. func sendErrMailApi(title, body string) {
  533. var tomail, api string
  534. if config.Conf.Mail.Send {
  535. tomail = config.Conf.Mail.To
  536. api = config.Conf.Mail.Api
  537. }
  538. log.Info("sendErrMailApi", zap.Any(tomail, api))
  539. res, err := http.Get(fmt.Sprintf("%s?to=%s&title=%s&body=%s", api, tomail, title, body))
  540. if err == nil {
  541. defer res.Body.Close()
  542. read, err := ioutil.ReadAll(res.Body)
  543. if err != nil {
  544. log.Info("邮件发送成功", zap.String("read", string(read)))
  545. }
  546. } else {
  547. log.Info("sendErrMailApi", zap.String("邮件发送失败", err.Error()))
  548. }
  549. }