Nats.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160
  1. package Nats
  2. import (
  3. "ColdVerify_server/Nats/NatsServer"
  4. "ColdVerify_server/conf"
  5. "ColdVerify_server/lib"
  6. "ColdVerify_server/logs"
  7. "ColdVerify_server/models/Device"
  8. "ColdVerify_server/models/System"
  9. "ColdVerify_server/models/Task"
  10. "fmt"
  11. "github.com/nats-io/nats.go"
  12. "github.com/vmihailenco/msgpack/v5"
  13. "strings"
  14. "sync"
  15. )
  16. func init() {
  17. logs.Println("============Nats init============")
  18. var err error
  19. // 连接Nats服务器
  20. lib.Nats, err = nats.Connect("nats://" + conf.NatsServer_Url)
  21. if err != nil {
  22. fmt.Println("nats 连接失败!")
  23. panic(err)
  24. }
  25. logs.Println("nats OK!")
  26. go NatsInit()
  27. }
  28. type Extract_TaskData_Back struct {
  29. T_uuid string `xml:"T_uuid"` // 任务主键id
  30. Time_start string `xml:"Time_start"`
  31. Time_end string `xml:"Time_end"`
  32. DeviceClassList []Device.DeviceClassList `xml:"DeviceClassList"` // 泛型
  33. Task Task.Task `xml:"Task"` // 泛型
  34. }
  35. func NatsInit() {
  36. // 发布-订阅 模式,打包数据
  37. _, _ = lib.Nats.Subscribe("ColdVerify_Server_Extract_TaskData_Back", func(m *nats.Msg) {
  38. logs.Debug("Extract_TaskData_Back message: %v\n", string(m.Data))
  39. var resp Extract_TaskData_Back
  40. err := msgpack.Unmarshal(m.Data, &resp)
  41. if err != nil {
  42. System.Add_Logs("Nats", "msgpack Unmarshal err", string(m.Data))
  43. return
  44. }
  45. Task_r := resp.Task
  46. // 清空表
  47. Task.Truncate_TaskData(Task_r.T_task_id)
  48. // 失败重试5次
  49. DeviceClassList := new(sync.Map)
  50. for _, v := range resp.DeviceClassList {
  51. DeviceClassList.Store(fmt.Sprintf("%s|%s", v.T_sn, v.T_id), 5)
  52. }
  53. DeviceClassList.Range(func(k, v any) bool {
  54. T_snid := strings.Split(k.(string), "|")
  55. T_sn := T_snid[0]
  56. T_id := T_snid[1]
  57. temp := v.(int)
  58. temp--
  59. DeviceClassList.Store(k, temp)
  60. err = Task.Import_TaskData_Back(T_sn, T_id, resp.Task.T_task_id, resp.Time_start, resp.Time_end)
  61. if err == nil || strings.Contains(err.Error(), "doesn't exist") {
  62. DeviceClassList.Delete(k)
  63. return true
  64. }
  65. return true
  66. })
  67. TaskData_Num := Task.Read_TaskData_Count(Task_r.T_task_id)
  68. //if TaskData_Num == 0 {
  69. // Task_r.T_collection_num = 0
  70. // if !Task.Update_Task(Task_r, "T_collection_state") {
  71. // logs.Error(lib.FuncName(), "后台执行修改任务数据失败")
  72. // }
  73. // return
  74. //}
  75. // 导入到本地数据
  76. NatsServer.Import_TaskData(resp.T_uuid, Task_r.T_task_id, TaskData_Num)
  77. System.Add_UserLogs_T(resp.T_uuid, "任务", "修改", Task_r)
  78. System.Add_UserLogs(resp.T_uuid, "提取数据", "提取数据"+Task_r.T_name, Task_r.T_task_id+"|"+resp.Time_start+"|"+resp.Time_end)
  79. })
  80. _, _ = lib.Nats.Subscribe("ColdVerify_Server_Update_Task", func(m *nats.Msg) {
  81. var t_Req Task.Task
  82. var t_R lib.JSONS
  83. err := msgpack.Unmarshal(m.Data, &t_Req)
  84. if err != nil {
  85. logs.Error("Mats", lib.FuncName(), err)
  86. t_R.Code = 202
  87. t_R.Msg = err.Error()
  88. b, _ := msgpack.Marshal(&t_R)
  89. _ = lib.Nats.Publish(m.Reply, b)
  90. return
  91. }
  92. fmt.Printf("ColdVerify_Server_Update_Task message: %+v\n", t_Req)
  93. col := []string{}
  94. if t_Req.T_delivery_state > 0 {
  95. col = append(col, "T_delivery_state")
  96. }
  97. if !Task.Update_Task(t_Req, "T_collection_state", "T_delivery_state") {
  98. logs.Error("Mats", lib.FuncName(), err)
  99. t_R.Code = 202
  100. t_R.Msg = err.Error()
  101. b, _ := msgpack.Marshal(&t_R)
  102. _ = lib.Nats.Publish(m.Reply, b)
  103. return
  104. }
  105. t_R.Code = 200
  106. t_R.Msg = "ok"
  107. b, _ := msgpack.Marshal(&t_R)
  108. _ = lib.Nats.Publish(m.Reply, b)
  109. })
  110. _, _ = lib.Nats.Subscribe("ColdVerify_Server_Read_Task", func(m *nats.Msg) {
  111. fmt.Printf("ColdVerify_Server_Read_Task message: %+v\n", string(m.Data))
  112. var t_R lib.JSONS
  113. task, is := Task.Read_Task(string(m.Data))
  114. if !is {
  115. logs.Error("Mats", lib.FuncName())
  116. t_R.Code = 202
  117. t_R.Msg = "查询失败"
  118. b, _ := msgpack.Marshal(&t_R)
  119. _ = lib.Nats.Publish(m.Reply, b)
  120. return
  121. }
  122. t_R.Code = 200
  123. t_R.Msg = "ok"
  124. t_R.Data = task
  125. b, _ := msgpack.Marshal(&t_R)
  126. _ = lib.Nats.Publish(m.Reply, b)
  127. })
  128. }