Procházet zdrojové kódy

保存文件数据丢失bug

xu před 2 týdny
rodič
revize
a6d91a9b87
9 změnil soubory, kde provedl 435 přidání a 151 odebrání
  1. 1 0
      dao/region.go
  2. 32 13
      initialize/myData.go
  3. 20 10
      modbus/operate.go
  4. 3 3
      model/common.go
  5. 16 11
      service/cron.go
  6. 49 31
      service/device.go
  7. 52 44
      static/data.json
  8. 0 12
      static/devicePath.json
  9. 262 27
      utils/myTool.go

+ 1 - 0
dao/region.go

@@ -5,4 +5,5 @@ type Region struct {
 	Name      string   `json:"name"`
 	SubRegion []Region `json:"subRegion"`
 	Devices   []Device `json:"devices"`
+	Version   int      `json:"version"` // 新增版本字段
 }

+ 32 - 13
initialize/myData.go

@@ -61,24 +61,29 @@ func StartInductanceTCP() {
 		logger.Get().Printf("lis Accept conn = %s\n", remoteAddr)
 
 		model.Mutex.Lock()
-		if _, exists := model.ConnectionMap[remoteAddr]; exists {
-			log.Printf("Connection from %s already exists, closing new connection\n", remoteAddr)
-			conn.Close()
-			model.Mutex.Unlock()
-			continue
-		}
+
 		deviceId, err := GetDeviceId(conn)
 		if err != nil {
 			log.Println("Error getting device ID:", err)
-			conn.Close()
+			//conn.Close()
 			model.Mutex.Unlock()
 			continue
 		}
-		model.ConnectionMap[deviceId] = conn
+
+		model.ConnectionMap1.Store(deviceId, conn)
+
 		model.Mutex.Unlock()
-		//用新的协程处理新的连接
-		go handler.ReadAndHandle(conn, deviceId)
-		go handler.Handler()
+		// 使用 Load 方法尝试获取连接
+		if conn1, ok := model.ConnectionMap1.Load(deviceId); ok {
+			// 成功找到连接
+			netConn := conn1.(net.Conn)
+			// 在这里处理 netConn
+			go handler.ReadAndHandle(netConn, deviceId)
+			go handler.Handler()
+		} else {
+			// 没有找到对应的连接
+			log.Printf("Connection for key %s not found", deviceId)
+		}
 	}
 }
 
@@ -240,10 +245,24 @@ func parseData(data *model.QueueData) {
 			//电池
 			if float64(batteryVoltage)/100 < 5 {
 				data1 := modbus.DeviceSwitch(8, 1)
-				utils.WriteDevice(data1, model.ConnectionMap[data.Id])
+				if conn1, ok := model.ConnectionMap1.Load(data.Id); ok {
+					// 成功找到连接
+					netConn := conn1.(net.Conn)
+					utils.WriteDevice(data1, netConn)
+				} else {
+					// 没有找到对应的连接
+					logger.Get().Printf("Connection for key %s not found", data.Id)
+				}
 			} else {
 				data1 := modbus.DeviceSwitch(8, 0)
-				utils.WriteDevice(data1, model.ConnectionMap[data.Id])
+				if conn1, ok := model.ConnectionMap1.Load(data.Id); ok {
+					// 成功找到连接
+					netConn := conn1.(net.Conn)
+					utils.WriteDevice(data1, netConn)
+				} else {
+					// 没有找到对应的连接
+					logger.Get().Printf("Connection for key %s not found", data.Id)
+				}
 			}
 		}
 	}

+ 20 - 10
modbus/operate.go

@@ -2,6 +2,7 @@ package modbus
 
 import (
 	"encoding/json"
+	"net"
 	"os"
 	"server/dao"
 	"server/logger"
@@ -12,24 +13,33 @@ import (
 
 // GetSunPowerInfo 获取太阳能信息
 func GetSunPowerInfo() error {
-	for id, conn := range model.ConnectionMap {
-		_, dev, err := utils.GetDataByDeviceId(id)
+	model.ConnectionMap1.Range(func(key, value interface{}) bool {
+		addr := key.(string)     // 假设键是 string 类型
+		conn := value.(net.Conn) // 假设值是 net.Conn 类型
+
+		_, dev, _ := utils.GetDataByDeviceId(addr)
 		if dev.IsSun && dev.State == 1 { //打开  并且在线
-			err = utils.WriteDevice(SolarEnergyData(), conn)
-			if err != nil {
-				return err
-			}
+			utils.WriteDevice(SolarEnergyData(), conn)
 		}
-	}
+
+		// 返回 true 继续遍历,返回 false 提前终止遍历
+		return true
+	})
 	return nil
 }
 
 // GetDeviceInfo 获取设备信息
 func GetDeviceInfo() {
-	for id, conn := range model.ConnectionMap {
-		_, dev, _ := utils.GetDataByDeviceId(id)
+	model.ConnectionMap1.Range(func(key, value interface{}) bool {
+		addr := key.(string)     // 假设键是 string 类型
+		conn := value.(net.Conn) // 假设值是 net.Conn 类型
+
+		_, dev, _ := utils.GetDataByDeviceId(addr)
 		utils.WriteDevice(ReadDeviceInfo(dev.LoopNumber), conn)
-	}
+
+		// 返回 true 继续遍历,返回 false 提前终止遍历
+		return true
+	})
 }
 
 // DealWithOffline 处理离线

+ 3 - 3
model/common.go

@@ -7,9 +7,9 @@ import (
 )
 
 var (
-	InductanceTCP net.Listener
-	ConnectionMap = make(map[string]net.Conn)
-	Mutex         sync.Mutex
+	InductanceTCP  net.Listener
+	Mutex          sync.Mutex
+	ConnectionMap1 = &sync.Map{}
 )
 
 type QueueData struct {

+ 16 - 11
service/cron.go

@@ -1,6 +1,7 @@
 package service
 
 import (
+	"net"
 	"server/logger"
 	"server/modbus"
 	"server/model"
@@ -106,19 +107,23 @@ func (c Cron) RelayOnOffTimeTask() {
 
 		for i, i2 := range value {
 			data := modbus.DeviceLoopSwitch(i, i2)
-			if model.ConnectionMap[key] == nil {
-				logger.Get().Errorf("设备连接丢失")
-			}
-			err := utils.WriteDevice(data, model.ConnectionMap[key])
-			time.Sleep(100 * time.Millisecond)
-			if err != nil {
-				logger.Get().Errorf("WriteDevice err = %s\n", err.Error())
-			}
+			if conn1, ok := model.ConnectionMap1.Load(key); ok {
+				// 成功找到连接
+				netConn := conn1.(net.Conn)
+				err := utils.WriteDevice(data, netConn)
+				time.Sleep(100 * time.Millisecond)
+				if err != nil {
+					logger.Get().Errorf("WriteDevice err = %s\n", err.Error())
+				}
 
-			for j, loop := range dev.DeviceLoops {
-				if loop.ID == i {
-					dev.DeviceLoops[j].State = i2
+				for j, loop := range dev.DeviceLoops {
+					if loop.ID == i {
+						dev.DeviceLoops[j].State = i2
+					}
 				}
+			} else {
+				// 没有找到对应的连接
+				logger.Get().Printf("Connection for key %s not found", key)
 			}
 		}
 

+ 49 - 31
service/device.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"fmt"
+	"net"
 	"server/dao"
 	"server/logger"
 	"server/modbus"
@@ -23,32 +24,38 @@ func SaveData(data []dao.Region) error {
 
 func DeviceLoopSwitch(deviceLoop dao.DeviceLoop) error {
 	data := modbus.DeviceLoopSwitch(deviceLoop.ID, deviceLoop.State)
-	logger.Get().Println(deviceLoop.DeviceId)
-	logger.Get().Println(model.ConnectionMap)
-	if model.ConnectionMap[deviceLoop.DeviceId] == nil {
-		return fmt.Errorf("设备连接丢失")
-	}
-	_, dev, _ := utils.GetDataByDeviceId(deviceLoop.DeviceId)
-	if dev.State == 0 {
-		return fmt.Errorf("设备离线无法操作")
-	}
+	if conn1, ok := model.ConnectionMap1.Load(deviceLoop.DeviceId); ok {
+		// 成功找到连接
+		netConn := conn1.(net.Conn)
+		_, dev, _ := utils.GetDataByDeviceId(deviceLoop.DeviceId)
+		if dev.State == 0 {
+			return fmt.Errorf("设备离线无法操作")
+		}
 
-	err := utils.WriteDevice(data, model.ConnectionMap[deviceLoop.DeviceId])
-	if err != nil {
-		fmt.Printf("Write device error: %v\n", err)
-		return err
+		err := utils.WriteDevice(data, netConn)
+		if err != nil {
+			fmt.Printf("Write device error: %v\n", err)
+			return err
+		}
+	} else {
+		// 没有找到对应的连接
+		logger.Get().Printf("Connection for key %s not found", deviceLoop.DeviceId)
 	}
 	return nil
 }
 func DeviceSwitch(req model.DeviceRequest) error {
 	data := modbus.DeviceSwitch(req.Device.LoopNumber, req.State)
-	if model.ConnectionMap[req.Device.Sn] == nil {
-		return fmt.Errorf("设备连接丢失")
-	}
-	err := utils.WriteDevice(data, model.ConnectionMap[req.Device.Sn])
-	if err != nil {
-		logger.Get().Errorf("Write device error: %v\n", err)
-		return err
+	if conn1, ok := model.ConnectionMap1.Load(req.Device.Sn); ok {
+		// 成功找到连接
+		netConn := conn1.(net.Conn)
+		err := utils.WriteDevice(data, netConn)
+		if err != nil {
+			logger.Get().Errorf("Write device error: %v\n", err)
+			return err
+		}
+	} else {
+		// 没有找到对应的连接
+		logger.Get().Printf("Connection for key %s not found", req.Device.Sn)
 	}
 	return nil
 }
@@ -59,13 +66,17 @@ func DeviceBatchSwitch(req model.DevicesRequest) error {
 			continue
 		}
 		data := modbus.DeviceSwitch(device.LoopNumber, req.State)
-		if model.ConnectionMap[device.Sn] == nil {
-			return fmt.Errorf("设备连接丢失")
-		}
-		err := utils.WriteDevice(data, model.ConnectionMap[device.Sn])
-		if err != nil {
-			fmt.Printf("Write device error: %v\n", err)
-			return err
+		if conn1, ok := model.ConnectionMap1.Load(device.Sn); ok {
+			// 成功找到连接
+			netConn := conn1.(net.Conn)
+			err := utils.WriteDevice(data, netConn)
+			if err != nil {
+				fmt.Printf("Write device error: %v\n", err)
+				return err
+			}
+		} else {
+			// 没有找到对应的连接
+			logger.Get().Printf("Connection for key %s not found", device.Sn)
 		}
 	}
 	return nil
@@ -100,10 +111,17 @@ func GetSunDevices() (devices []dao.Device, err error) {
 }
 
 func UpdateRegisterAddress(deviceId string) error {
-	err := utils.WriteDevice(modbus.UpdateRegisterAddress(), model.ConnectionMap[deviceId])
-	if err != nil {
-		logger.Get().Errorf("UpdateRegisterAddress error: %v\n", err)
-		return err
+	if conn1, ok := model.ConnectionMap1.Load(deviceId); ok {
+		// 成功找到连接
+		netConn := conn1.(net.Conn)
+		err := utils.WriteDevice(modbus.UpdateRegisterAddress(), netConn)
+		if err != nil {
+			logger.Get().Errorf("UpdateRegisterAddress error: %v\n", err)
+			return err
+		}
+	} else {
+		// 没有找到对应的连接
+		logger.Get().Printf("Connection for key %s not found", deviceId)
 	}
 	return nil
 }

+ 52 - 44
static/data.json

@@ -1,4 +1,25 @@
 [
+  {
+    "id": 6,
+    "name": "搅拌站至三岔路口",
+    "subRegion": null,
+    "devices": [],
+    "version": 136
+  },
+  {
+    "id": 7,
+    "name": "垃圾堆体道路",
+    "subRegion": null,
+    "devices": [],
+    "version": 136
+  },
+  {
+    "id": 8,
+    "name": "污水进场道路",
+    "subRegion": null,
+    "devices": [],
+    "version": 136
+  },
   {
     "id": 1,
     "name": "生活区",
@@ -13,15 +34,15 @@
         "state": 1,
         "isSun": true,
         "loopNumber": 8,
-        "onlineTime": "2025-01-19T09:43:59.91502361+08:00",
+        "onlineTime": "2025-03-19T13:40:00.0099855+08:00",
         "deviceLoops": [
           {
             "id": 1,
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路12",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -30,8 +51,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路2",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -40,8 +61,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路3",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -50,8 +71,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路4",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -60,8 +81,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路5",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -70,8 +91,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路6",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -80,8 +101,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路7",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           },
@@ -90,8 +111,8 @@
             "deviceId": "JM36xWRZq6PiwmKV",
             "name": "回路8",
             "state": 0,
-            "timeCondition1OnTime": "关闭",
-            "timeCondition1OffTime": "关闭",
+            "timeCondition1OnTime": "00:50",
+            "timeCondition1OffTime": "08:50",
             "timeCondition2OnTime": "关闭",
             "timeCondition2OffTime": "关闭"
           }
@@ -109,7 +130,7 @@
         "regionId": 1,
         "name": "洗车房",
         "genre": "四回路控制",
-        "state": 1,
+        "state": 0,
         "isSun": false,
         "loopNumber": 4,
         "onlineTime": "2025-01-19T09:46:00.028035846+08:00",
@@ -168,7 +189,7 @@
         "regionId": 1,
         "name": "污泥厂大配电房",
         "genre": "四回路控制",
-        "state": 1,
+        "state": 0,
         "isSun": false,
         "loopNumber": 4,
         "onlineTime": "2025-01-19T09:46:00.027293637+08:00",
@@ -227,7 +248,7 @@
         "regionId": 1,
         "name": "生活区",
         "genre": "四回路控制",
-        "state": 1,
+        "state": 0,
         "isSun": false,
         "loopNumber": 4,
         "onlineTime": "2025-01-19T09:44:00.051568026+08:00",
@@ -286,7 +307,7 @@
         "regionId": 1,
         "name": "入厂道路",
         "genre": "八回路控制",
-        "state": 1,
+        "state": 0,
         "isSun": false,
         "loopNumber": 8,
         "onlineTime": "2025-01-19T09:46:00.039018537+08:00",
@@ -379,48 +400,35 @@
           "batteryPlateVoltage": 0
         }
       }
-    ]
+    ],
+    "version": 136
   },
   {
     "id": 2,
     "name": "股份公司进场上坡",
     "subRegion": null,
-    "devices": []
+    "devices": [],
+    "version": 136
   },
   {
     "id": 3,
     "name": "地磅房至污泥厂门口",
     "subRegion": null,
-    "devices": []
+    "devices": [],
+    "version": 136
   },
   {
     "id": 4,
     "name": "三角花园至油库",
     "subRegion": null,
-    "devices": []
+    "devices": [],
+    "version": 136
   },
   {
     "id": 5,
     "name": "地磅房至搅拌站",
     "subRegion": null,
-    "devices": []
-  },
-  {
-    "id": 6,
-    "name": "搅拌站至三岔路口",
-    "subRegion": null,
-    "devices": []
-  },
-  {
-    "id": 7,
-    "name": "垃圾堆体道路",
-    "subRegion": null,
-    "devices": []
-  },
-  {
-    "id": 8,
-    "name": "污水进场道路",
-    "subRegion": null,
-    "devices": []
+    "devices": [],
+    "version": 136
   }
 ]

+ 0 - 12
static/devicePath.json

@@ -1,12 +0,0 @@
-[
-  {
-    "path": "192.168.110.212:2",
-    "id": "JM032347132tWa6g",
-    "tcp": null
-  },
-  {
-    "path": "192.168.110.214:2",
-    "id": "JM03g8PcVDVxLNHn",
-    "tcp": null
-  }
-]

+ 262 - 27
utils/myTool.go

@@ -17,31 +17,22 @@ const dataFilePath = "static/data.json"
 
 // LoadData 从文件中加载 Region 数据。
 func LoadData() ([]dao.Region, error) {
-	data, err := os.ReadFile(dataFilePath)
-
-	datafb, err := os.ReadFile("static/data-fb.json")
-	var regionsfb []dao.Region
-	if err := json.Unmarshal(datafb, &regionsfb); err != nil {
-		// 提供更多的上下文信息来帮助调试
-		return nil, fmt.Errorf("解析 JSON 失败: %v, 原始数据: %s", err, string(data))
-	}
+	var regions []dao.Region
 
+	// 尝试从主数据文件读取数据
+	data, err := os.ReadFile(dataFilePath)
 	if err != nil {
-		if os.IsNotExist(err) {
-			return regionsfb, nil // 文件不存在时返回空数组
+		if !os.IsNotExist(err) {
+			return nil, fmt.Errorf("读取文件失败: %v", err)
+		}
+		// 如果主数据文件不存在,则尝试从备份文件读取
+	} else {
+		// 检查文件是否为空
+		if len(data) > 0 {
+			if err := json.Unmarshal(data, &regions); err != nil {
+				return nil, fmt.Errorf("解析 JSON 失败 (主文件): %v, 原始数据: %s", err, string(data))
+			}
 		}
-		return nil, fmt.Errorf("读取文件失败: %v", err)
-	}
-
-	// 检查文件是否为空
-	if len(data) == 0 {
-		return regionsfb, nil // 如果文件为空,返回空数组
-	}
-
-	var regions []dao.Region
-	if err := json.Unmarshal(data, &regions); err != nil {
-		// 提供更多的上下文信息来帮助调试
-		return nil, fmt.Errorf("解析 JSON 失败: %v, 原始数据: %s", err, string(data))
 	}
 
 	return regions, nil
@@ -103,23 +94,56 @@ func SaveRegionOnData(data dao.Region) ([]dao.Region, error) {
 	if err != nil {
 		return regions, err
 	}
+
 	for i, region := range regions {
 		if region.Name == data.Name {
 			regions[i] = data
+			break
 		}
 	}
+
 	return regions, nil
 }
 
 // SaveData 保存数据到文件
-func SaveData(path string, parameter interface{}) error {
-	// 如果 parameter 是 nil 或者是空的结构体/切片,可以选择不写入文件或返回错误。
-	if parameter == nil {
-		return fmt.Errorf("无法保存 nil 数据")
+func SaveData(path string, newData []dao.Region) error {
+	// 加载当前数据
+	currentRegions, err := LoadData()
+	if err != nil {
+		return err
+	}
+
+	// 创建一个映射以便快速查找
+	regionMap := make(map[string]dao.Region)
+	for _, region := range currentRegions {
+		regionMap[region.Name] = region
+	}
+
+	// 遍历传入的新数据,进行版本控制检查
+	for _, newRegion := range newData {
+		existingRegion, exists := regionMap[newRegion.Name]
+		if exists {
+			if existingRegion.Version != newRegion.Version {
+				return fmt.Errorf("版本冲突: 区域 '%s' 的当前版本=%d, 提交版本=%d", newRegion.Name, existingRegion.Version, newRegion.Version)
+			}
+			// 更新版本号
+			newRegion.Version++
+			regionMap[newRegion.Name] = newRegion
+		} else {
+			// 如果区域不存在,则直接添加(假设这是新增的区域)
+			newRegion.Version = 1 // 新增区域的初始版本号设为1
+			regionMap[newRegion.Name] = newRegion
+		}
+	}
+
+	// 将 map 转换回 slice
+	updatedRegions := make([]dao.Region, 0, len(regionMap))
+	for _, region := range regionMap {
+		updatedRegions = append(updatedRegions, region)
 	}
 
 	// 序列化数据
-	data, err := json.MarshalIndent(parameter, "", "  ")
+	data, err := json.MarshalIndent(updatedRegions, "", "  ")
 	if err != nil {
 		return fmt.Errorf("JSON 序列化失败: %v", err)
 	}
@@ -158,10 +182,221 @@ func SaveData(path string, parameter interface{}) error {
 
 const (
 	maxRetries    = 3               // 最大重试次数
+	readTimeout   = 5 * time.Second // 读取超时时间
 	writeTimeout  = 5 * time.Second // 写入超时时间
 	reconnectWait = 2 * time.Second // 重连等待时间
 )
 
+// checkConnection 检查连接是否仍然有效
+func checkConnection(conn net.Conn) error {
+	probe := []byte{}
+	_, err := conn.Write(probe)
+	return err
+}
+
+// checkAndReconnect 检查是否需要重连,并在必要时执行重连
+func checkAndReconnect(conn net.Conn) (net.Conn, error) {
+	remoteAddr := conn.RemoteAddr().String()
+
+	// 关闭旧连接
+	if err := conn.Close(); err != nil {
+		logger.Get().Errorf("Failed to close connection: %v", err)
+		return nil, err
+	}
+
+	// 解析原始连接的远程地址和网络接口
+	addr, networkInterface, err := parseRemoteAddr(remoteAddr)
+	if err != nil {
+		logger.Get().Errorf("Failed to parse remote address: %v", err)
+		return nil, err
+	}
+
+	// 尝试重新建立连接
+	newConn, err := net.Dial("tcp", fmt.Sprintf("%s%%%s", addr, networkInterface))
+	if err != nil {
+		logger.Get().Errorf("Reconnect failed: %v", err)
+		return nil, err
+	}
+
+	logger.Get().Warnf("Successfully reconnected to %s", remoteAddr)
+	return newConn, nil
+}
+
+// isConnectionClosedError 检查错误是否表明连接被对端强制关闭或已关闭
+func isConnectionClosedError(err error) bool {
+	if ne, ok := err.(*net.OpError); ok {
+		return strings.Contains(ne.Err.Error(), "forcibly closed") ||
+			strings.Contains(ne.Err.Error(), "broken pipe") ||
+			strings.Contains(ne.Err.Error(), "connection reset") ||
+			strings.Contains(ne.Err.Error(), "use of closed network connection")
+	}
+	return false
+}
+
+// setDeadlineWithRetry 使用重试机制设置读写截止时间
+func setDeadlineWithRetry(conn net.Conn, timeout time.Duration, operation string) error {
+	for attempts := 0; attempts < maxRetries; attempts++ {
+		var err error
+		switch operation {
+		case "read":
+			err = conn.SetReadDeadline(time.Now().Add(timeout))
+		case "write":
+			err = conn.SetWriteDeadline(time.Now().Add(timeout))
+		default:
+			return fmt.Errorf("invalid operation: %s", operation)
+		}
+
+		if err == nil {
+			return nil
+		}
+
+		if isConnectionClosedError(err) {
+			logger.Get().Warnf("Set %s deadline failed due to closed connection, retrying (%d/%d)", operation, attempts+1, maxRetries)
+
+			var newConn net.Conn
+			var reconnErr error
+
+			newConn, reconnErr = checkAndReconnect(conn)
+			if reconnErr != nil {
+				time.Sleep(reconnectWait) // 等待一段时间后重试
+				continue                  // 继续下一次重试
+			}
+
+			conn = newConn
+			continue // 重试设置截止时间
+		}
+
+		logger.Get().Errorf("Set %s deadline failed: %v", operation, err)
+		return fmt.Errorf("set %s deadline failed after %d retries: %v", operation, attempts+1, err)
+	}
+
+	return fmt.Errorf("failed to set %s deadline after %d retries", operation, maxRetries)
+}
+
+// readWithRetry 使用重试机制进行读取操作
+func readWithRetry(buffer []byte, conn net.Conn) (int, error) {
+	for attempts := 0; attempts < maxRetries; attempts++ {
+		if err := setDeadlineWithRetry(conn, readTimeout, "read"); err != nil {
+			return 0, err
+		}
+
+		// 检查连接是否仍然有效
+		if err := checkConnection(conn); err != nil {
+			if isConnectionClosedError(err) {
+				logger.Get().Warnf("Connection check failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries)
+
+				var newConn net.Conn
+				var reconnErr error
+
+				newConn, reconnErr = checkAndReconnect(conn)
+				if reconnErr != nil {
+					time.Sleep(reconnectWait) // 等待一段时间后重试
+					continue                  // 继续下一次重试
+				}
+
+				conn = newConn
+				continue // 重试读取
+			}
+
+			logger.Get().Errorf("Connection check failed: %v", err)
+			return 0, fmt.Errorf("connection check failed after %d retries: %v", attempts+1, err)
+		}
+
+		n, err := conn.Read(buffer)
+		if err == nil {
+			return n, nil
+		}
+
+		if isConnectionClosedError(err) {
+			logger.Get().Warnf("Read failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries)
+
+			var newConn net.Conn
+			var reconnErr error
+
+			newConn, reconnErr = checkAndReconnect(conn)
+			if reconnErr != nil {
+				time.Sleep(reconnectWait) // 等待一段时间后重试
+				continue                  // 继续下一次重试
+			}
+
+			conn = newConn
+			continue // 重试读取
+		}
+
+		logger.Get().Errorf("Read failed: %v", err)
+		return 0, fmt.Errorf("read failed after %d retries: %v", attempts+1, err)
+	}
+
+	return 0, fmt.Errorf("failed to read after %d retries", maxRetries)
+}
+
+// writeWithRetry 使用重试机制进行写入操作
+func writeWithRetry(frame []byte, conn net.Conn) error {
+	for attempts := 0; attempts < maxRetries; attempts++ {
+		if err := setDeadlineWithRetry(conn, writeTimeout, "write"); err != nil {
+			return err
+		}
+
+		// 检查连接是否仍然有效
+		if err := checkConnection(conn); err != nil {
+			if isConnectionClosedError(err) {
+				logger.Get().Warnf("Connection check failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries)
+
+				var newConn net.Conn
+				var reconnErr error
+
+				newConn, reconnErr = checkAndReconnect(conn)
+				if reconnErr != nil {
+					time.Sleep(reconnectWait) // 等待一段时间后重试
+					continue                  // 继续下一次重试
+				}
+
+				conn = newConn
+				continue // 重试写入
+			}
+
+			logger.Get().Errorf("Connection check failed: %v", err)
+			return fmt.Errorf("connection check failed after %d retries: %v", attempts+1, err)
+		}
+
+		_, err := conn.Write(frame)
+		if err == nil {
+			return nil
+		}
+
+		if isConnectionClosedError(err) {
+			logger.Get().Warnf("Write failed due to closed connection, retrying (%d/%d)", attempts+1, maxRetries)
+
+			var newConn net.Conn
+			var reconnErr error
+
+			newConn, reconnErr = checkAndReconnect(conn)
+			if reconnErr != nil {
+				time.Sleep(reconnectWait) // 等待一段时间后重试
+				continue                  // 继续下一次重试
+			}
+
+			conn = newConn
+			continue // 重试写入
+		}
+
+		logger.Get().Errorf("Write failed: %v", err)
+		return fmt.Errorf("write failed after %d retries: %v", attempts+1, err)
+	}
+
+	return fmt.Errorf("failed to write after %d retries", maxRetries)
+}
+
+// ReadDevice 从设备读取数据
+func ReadDevice(buffer []byte, conn net.Conn) (int, error) {
+	return readWithRetry(buffer, conn)
+}
+
+// WriteDevice1 向设备写入数据
+func WriteDevice1(frame []byte, conn net.Conn) error {
+	return writeWithRetry(frame, conn)
+}
+
 func WriteDevice(frame []byte, conn net.Conn) error {
 	for attempts := 0; attempts < maxRetries; attempts++ {
 		if err := conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil {