myData.go 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290
  1. package initialize
  2. import (
  3. "encoding/hex"
  4. "net"
  5. "os"
  6. "runtime"
  7. "runtime/debug"
  8. "server/dao"
  9. "server/logger"
  10. "server/modbus"
  11. "server/model"
  12. "server/service"
  13. "server/utils"
  14. "strconv"
  15. "strings"
  16. "sync"
  17. "syscall"
  18. "time"
  19. )
  20. type ModbusHandler struct {
  21. queue *modbus.MlQueue
  22. }
  23. var _handlerOnce sync.Once
  24. var _handlerSingle *ModbusHandler
  25. func GetHandler() *ModbusHandler {
  26. _handlerOnce.Do(func() {
  27. _handlerSingle = &ModbusHandler{
  28. queue: modbus.NewQueue(10000),
  29. }
  30. })
  31. return _handlerSingle
  32. }
  33. func InitInductanceTCP() {
  34. lis, err := net.Listen("tcp", ":60001")
  35. if err != nil {
  36. logger.Get().Fatalln("Failed to listen on port 60001:", err)
  37. } else {
  38. logger.Get().Println("inductanceTCP启动成功")
  39. model.InductanceTCP = lis
  40. }
  41. go StartInductanceTCP()
  42. }
  43. func StartInductanceTCP() {
  44. logger.Get().Println("进入StartInductanceTCP")
  45. handler := GetHandler()
  46. for {
  47. logger.Get().Println("进入InductanceTCP循环")
  48. conn, err := model.InductanceTCP.Accept()
  49. if err != nil {
  50. logger.Get().Errorf("Accept error: %v", err)
  51. // 根据实际情况决定是否继续循环
  52. continue
  53. }
  54. remoteAddr := conn.RemoteAddr().String()
  55. logger.Get().Printf("lis Accept conn = %s\n", remoteAddr)
  56. deviceId, err := GetDeviceId(conn)
  57. if err != nil {
  58. logger.Get().Errorf("Error getting device ID from %s: %v", remoteAddr, err)
  59. conn.Close()
  60. continue
  61. }
  62. if len(deviceId) != 16 {
  63. continue
  64. }
  65. model.Mutex.Lock()
  66. if existingConn, exists := model.ConnectionMap[deviceId]; exists {
  67. logger.Get().Printf("Connection from %s already exists for deviceId %d, closing new connection\n", remoteAddr, deviceId)
  68. existingConn.Close() // 关闭旧连接
  69. delete(model.ConnectionMap, deviceId) // 清理旧连接
  70. }
  71. model.ConnectionMap[deviceId] = conn
  72. logger.Get().Printf("lis Accept conn = %s, 添加进入 %s \n", remoteAddr, deviceId)
  73. model.Mutex.Unlock()
  74. // 使用新的协程处理新的连接
  75. go handler.ReadAndHandle(conn, deviceId)
  76. go handler.Handler()
  77. }
  78. }
  79. func GetDeviceId(conn net.Conn) (id string, err error) {
  80. // 发送 Modbus RTU 帧
  81. //FE 04 03 EE 00 08 85 B2
  82. data, err := utils.WriteAndReadDevice(modbus.ReadDeviceId, conn, 9, 2)
  83. if err != nil {
  84. return
  85. }
  86. return string(data), err
  87. }
  88. func (o *ModbusHandler) ReadAndHandle(conn net.Conn, deviceId string) {
  89. for {
  90. buffer := make([]byte, 1024)
  91. n, err := conn.Read(buffer)
  92. if err != nil {
  93. if isConnReset(err) {
  94. logger.Get().Error("连接被远程主机强制关闭")
  95. } else if os.IsTimeout(err) {
  96. logger.Get().Error("读取操作超时")
  97. } else {
  98. // 处理其他类型的错误
  99. logger.Get().Errorf("读取错误: %s\n", err)
  100. }
  101. return
  102. }
  103. queueData := model.QueueData{
  104. Id: deviceId,
  105. Value: buffer[:n],
  106. }
  107. ok, cnt := o.queue.Put(&queueData)
  108. if ok {
  109. continue
  110. } else {
  111. logger.Get().Errorf("HandlerData:查询队列失败,队列消息数量:%d", cnt)
  112. runtime.Gosched()
  113. }
  114. }
  115. }
  116. func isConnReset(err error) bool {
  117. if opErr, ok := err.(*net.OpError); ok {
  118. if opErr.Err == syscall.ECONNRESET {
  119. return true // Unix-like 系统上的 ECONNRESET
  120. } else if runtime.GOOS == "windows" {
  121. // Windows 上的 WSAECONNRESET 通常是通过错误消息识别的
  122. if se, ok := opErr.Err.(*os.SyscallError); ok {
  123. if errno, ok := se.Err.(syscall.Errno); ok {
  124. if errno == 10054 { // 10054 对应 WSAECONNRESET
  125. return true
  126. }
  127. }
  128. } else if strings.Contains(opErr.Err.Error(), "WSAECONNRESET") {
  129. // 如果错误消息包含 WSAECONNRESET,也认为是连接被重置
  130. return true
  131. }
  132. }
  133. }
  134. return false
  135. }
  136. func (o *ModbusHandler) Handler() interface{} {
  137. defer func() {
  138. if err := recover(); err != nil {
  139. go GetHandler().Handler()
  140. logger.Get().Errorf("MqttHandler.Handler:发生异常:%s", string(debug.Stack()))
  141. }
  142. }()
  143. for {
  144. msg, ok, quantity := o.queue.Get()
  145. if !ok {
  146. time.Sleep(10 * time.Millisecond)
  147. continue
  148. } else if quantity > 1000 {
  149. logger.Get().Errorf("数据队列累积过多,请注意优化,当前队列条数:%d", quantity)
  150. }
  151. queueData, ok := msg.(*model.QueueData)
  152. if !ok {
  153. logger.Get().Errorln("Type assertion failed: msg is not of type model.QueueDat")
  154. return nil
  155. }
  156. // 信息处理返回
  157. parseData(queueData)
  158. // 对数据进行修改
  159. }
  160. }
  161. func parseData(data *model.QueueData) {
  162. reg, dev, err := utils.GetDataByDeviceId(data.Id)
  163. if err != nil {
  164. logger.Get().Errorln("Error getting register and device:", err)
  165. return
  166. }
  167. toString := hex.EncodeToString(data.Value)
  168. switch toString[0:2] {
  169. case "fe":
  170. switch toString[4:8] { // 开关灯
  171. case "0000", "0001", "0002", "0003", "0004", "0005", "0006", "0007":
  172. logger.Get().Println(dev.Sn + "-----" + toString)
  173. zhi, _ := strconv.Atoi(toString[7:8])
  174. if toString[8:12] == "0000" {
  175. dev.DeviceLoops[zhi].State = 0
  176. } else if toString[8:12] == "ff00" {
  177. dev.DeviceLoops[zhi].State = 1
  178. }
  179. for i, device := range reg.Devices {
  180. if device.Sn == data.Id {
  181. reg.Devices[i] = dev
  182. }
  183. }
  184. regions, err := utils.SaveRegionOnData(reg)
  185. err = service.SaveData(regions)
  186. if err != nil {
  187. logger.Get().Errorln("设备回路状态" + err.Error())
  188. return
  189. }
  190. }
  191. switch toString[2:6] {
  192. case "0101":
  193. for i, device := range reg.Devices {
  194. if device.Sn == data.Id {
  195. if reg.Devices[i].State != 1 {
  196. reg.Devices[i].State = 1
  197. service.Cron{}.RelayOnOffTimeTaskSn(data.Id)
  198. }
  199. reg.Devices[i].OnlineTime = time.Now()
  200. two, err := utils.SixteenTurnsTwo(toString[6:8])
  201. if err != nil {
  202. logger.Get().Errorln("16转2进制" + err.Error())
  203. return
  204. }
  205. for i2, _ := range reg.Devices[i].DeviceLoops {
  206. state := string(two[7-i2])
  207. zhi, _ := strconv.Atoi(state)
  208. reg.Devices[i].DeviceLoops[i2].State = zhi
  209. }
  210. }
  211. }
  212. regions, err := utils.SaveRegionOnData(reg)
  213. err = service.SaveData(regions)
  214. if err != nil {
  215. logger.Get().Errorln("设备回路状态" + err.Error())
  216. return
  217. }
  218. }
  219. case "01":
  220. switch toString[2:6] {
  221. case "0336":
  222. batteryVoltage, _ := strconv.ParseInt(toString[6:10], 16, 64)
  223. batteryCurrent, _ := strconv.ParseInt(toString[10:14], 16, 64)
  224. batteryPlateVoltage, _ := strconv.ParseInt(toString[38:42], 16, 64)
  225. sun := dao.Sun{
  226. DeviceId: dev.Sn,
  227. BatteryVoltage: float64(batteryVoltage) / 100,
  228. BatteryCurrent: int(batteryCurrent),
  229. BatteryPlateVoltage: float64(batteryPlateVoltage) / 100,
  230. }
  231. for i, device := range reg.Devices {
  232. if device.Sn == dev.Sn {
  233. reg.Devices[i].Sun = sun
  234. }
  235. }
  236. regions, err := utils.SaveRegionOnData(reg)
  237. err = service.SaveData(regions)
  238. if err != nil {
  239. logger.Get().Println("太阳能数据保存" + err.Error())
  240. return
  241. }
  242. //电池
  243. //if float64(batteryVoltage)/100 < 5 {
  244. // data1 := modbus.DeviceSwitch(8, 1)
  245. // utils.WriteDevice(data1, model.ConnectionMap[data.Id])
  246. //} else {
  247. // data1 := modbus.DeviceSwitch(8, 0)
  248. // utils.WriteDevice(data1, model.ConnectionMap[data.Id])
  249. //}
  250. }
  251. }
  252. switch toString[0:4] {
  253. case "4c43":
  254. bytes, err := hex.DecodeString(toString[4:])
  255. if err != nil {
  256. logger.Get().Errorln("Error decoding bytes:", err)
  257. return
  258. }
  259. for i, device := range reg.Devices {
  260. if device.Sn == string(bytes) {
  261. reg.Devices[i].State = 1
  262. reg.Devices[i].OnlineTime = time.Now()
  263. }
  264. }
  265. regions, err := utils.SaveRegionOnData(reg)
  266. err = service.SaveData(regions)
  267. if err != nil {
  268. logger.Get().Errorln("心跳" + err.Error())
  269. return
  270. }
  271. }
  272. }