main.go 5.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
  1. package main
  2. import (
  3. "encoding/json"
  4. "fmt"
  5. "go.uber.org/zap"
  6. . "jy_publishing/Logger"
  7. ms "jy_publishing/megaloscope"
  8. nsq "jy_publishing/nsq"
  9. "jy_publishing/tool"
  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/mongodb"
  13. "jygit.jydev.jianyu360.cn/data_processing/common_utils/udp"
  14. "log"
  15. "net"
  16. )
  17. var (
  18. Itype string
  19. MCJy, MCAtts *nsq.Consumer
  20. )
  21. func init() {
  22. InitLogger("./log/All.log", "debug")
  23. util.ReadConfig(&tool.Sysconfig)
  24. //信息类型
  25. if tool.Sysconfig["infoCode"] != nil {
  26. b, err := json.Marshal(tool.Sysconfig["infoCode"])
  27. if err == nil {
  28. err = json.Unmarshal(b, &tool.InfoCodes)
  29. }
  30. if err != nil {
  31. log.Println("infoCode init err :", err)
  32. }
  33. }
  34. bidding := tool.Sysconfig["bidding"].(map[string]interface{})
  35. tool.BidColl = bidding["dbColl"].(string)
  36. tool.MgoBid = &mongodb.MongodbSim{
  37. MongodbAddr: bidding["addr"].(string),
  38. Size: util.IntAllDef(tool.Sysconfig["mgoPoolSize"], 5),
  39. DbName: bidding["dbName"].(string),
  40. UserName: bidding["uname"].(string),
  41. Password: bidding["upwd"].(string),
  42. }
  43. tool.MgoBid.InitPool()
  44. extract := tool.Sysconfig["extract"].(map[string]interface{})
  45. tool.ExtColl = extract["dbColl"].(string)
  46. tool.ExtColl1 = extract["dbColl1"].(string)
  47. tool.MgoExt = &mongodb.MongodbSim{
  48. MongodbAddr: extract["addr"].(string),
  49. Size: util.IntAllDef(tool.Sysconfig["mgoPoolSize"], 5),
  50. DbName: extract["dbName"].(string),
  51. }
  52. tool.MgoExt.InitPool()
  53. es := tool.Sysconfig["es"].(map[string]interface{})
  54. if util.ObjToString(es["addr"]) != "" {
  55. tool.Es = &elastic.Elastic{
  56. S_esurl: util.ObjToString(es["addr"]),
  57. I_size: util.IntAllDef(es["pool"], 10),
  58. Username: util.ObjToString(es["user"]),
  59. Password: util.ObjToString(es["password"]),
  60. }
  61. tool.Index = util.ObjToString(es["index"])
  62. tool.IndexAll = util.ObjToString(es["index_all"])
  63. Itype = util.ObjToString(es["itype"])
  64. tool.Es.InitElasticSize()
  65. log.Println("初始化 elasticsearch")
  66. }
  67. //加载敏感词文件
  68. tool.Ms = ms.NewMegaloscope("./rules.txt")
  69. tool.InitOss()
  70. tool.InitEtcd()
  71. initUdp()
  72. tool.Oss_Server_Address = tool.Sysconfig["ossServerAds"].(string)
  73. }
  74. func initUdp() {
  75. updport := tool.Sysconfig["udpPort"].(string)
  76. tool.UdpClient = udp.UdpClient{Local: updport, BufSize: 1024}
  77. Logger.Info("Udp 监听 port: " + updport)
  78. tool.UdpClient.Listen(processUdpMsg)
  79. }
  80. func main() {
  81. go jyNsqMethod()
  82. go attsNsqMethod()
  83. c := make(chan bool, 1)
  84. <-c
  85. }
  86. // @Description 剑鱼消息队列 按照类型处理消息
  87. // @Author J 2022/4/14 11:42 AM
  88. func jyNsqMethod() {
  89. cof := tool.Sysconfig["nsq_jy"].(map[string]interface{})
  90. var err error
  91. MCJy, err = nsq.NewConsumer(&nsq.Cconfig{
  92. IsJsonEncode: true, //与生产者配置对应,设为true会取第1个字节进行类型判断
  93. Addr: util.ObjToString(cof["addr"]),
  94. ConnectType: 1, //默认连接nsqd
  95. Topic: util.ObjToString(cof["topic"]),
  96. Channel: util.ObjToString(cof["channel"]),
  97. Concurrent: util.IntAllDef(cof["concurrent"], 1), //并发数
  98. })
  99. if err != nil {
  100. Logger.Error(err.Error())
  101. }
  102. for {
  103. select {
  104. case obj := <-MCJy.Ch: //从通道读取即可
  105. Logger.Info("jy nsq: " + fmt.Sprint(obj))
  106. tool.TaskInfo(obj)
  107. }
  108. }
  109. }
  110. // @Description 附件处理完成的数据
  111. // @Author J 2022/4/14 1:18 PM
  112. func attsNsqMethod() {
  113. var err error
  114. cofAtts := tool.Sysconfig["nsq_attachment"].(map[string]interface{})
  115. tool.MProducer, err = nsq.NewProducer(util.ObjToString(cofAtts["addr_p"]), util.ObjToString(cofAtts["topic"]), true)
  116. if err != nil {
  117. Logger.Error(err.Error())
  118. }
  119. tool.FileTopicResult = util.ObjToString(cofAtts["topic-result"])
  120. MCAtts, err = nsq.NewConsumer(&nsq.Cconfig{
  121. IsJsonEncode: true,
  122. Addr: util.ObjToString(cofAtts["addr_c"]),
  123. ConnectType: 1, //默认连接nsqd
  124. Topic: tool.FileTopicResult,
  125. Channel: util.ObjToString(cofAtts["channel"]),
  126. Concurrent: util.IntAllDef(cofAtts["concurrent"], 1), //并发数
  127. })
  128. if err != nil {
  129. Logger.Error(err.Error())
  130. }
  131. for {
  132. select {
  133. case obj := <-MCAtts.Ch:
  134. Logger.Info("file extract receive nsq: " + fmt.Sprint(obj))
  135. tool.TaskAtts(obj.(map[string]interface{}))
  136. }
  137. }
  138. }
  139. func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
  140. switch act {
  141. case udp.OP_TYPE_DATA: //测试接收
  142. var resp map[string]interface{}
  143. err := json.Unmarshal(data, &resp)
  144. Logger.Info("udp receive...", zap.Any("data", resp))
  145. if err != nil {
  146. //go Udpclient.WriteUdp([]byte{}, mu.OP_NOOP, ra) //回应上一个节点
  147. } else if resp != nil {
  148. key, _ := resp["key"].(string)
  149. if key == "" {
  150. key = "udpok"
  151. }
  152. go tool.UdpClient.WriteUdp([]byte(key), udp.OP_NOOP, ra)
  153. tasktype, _ := resp["stype"].(string)
  154. switch tasktype {
  155. case "jyfb_data_over":
  156. go func() {
  157. tool.JyRpcDataFin(resp["infoid"].(string))
  158. }()
  159. }
  160. }
  161. case udp.OP_NOOP: //下个节点回应
  162. Logger.Info("接收回应:", zap.Any("data", string(data)))
  163. }
  164. }