package utils import ( "fmt" "net" "server/global" "server/model" "strings" "time" ) 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() // 解析远程地址 addr1, err := net.ResolveTCPAddr("tcp", remoteAddr) if err != nil { // 处理错误... global.GVA_LOG.Error(fmt.Sprintf("解析错误 conn = %s\n", addr1.IP.String())) } // 关闭旧连接 if err := conn.Close(); err != nil { global.GVA_LOG.Error(fmt.Sprintf("Failed to close connection: %v", err)) return nil, err } // 解析原始连接的远程地址和网络接口 addr, networkInterface, err := parseRemoteAddr(addr1.IP.String()) if err != nil { global.GVA_LOG.Error(fmt.Sprintf("Failed to parse remote address: %v", err)) return nil, err } // 尝试重新建立连接 newConn, err := net.Dial("tcp", fmt.Sprintf("%s%%%s", addr, networkInterface)) if err != nil { global.GVA_LOG.Error(fmt.Sprintf("Reconnect failed: %v", err)) return nil, err } 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) { global.GVA_LOG.Warn(fmt.Sprintf("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 // 重试设置截止时间 } 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) { global.GVA_LOG.Warn(fmt.Sprintf("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 // 重试读取 } global.GVA_LOG.Warn(fmt.Sprintf("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) { var newConn net.Conn var reconnErr error newConn, reconnErr = checkAndReconnect(conn) if reconnErr != nil { time.Sleep(reconnectWait) // 等待一段时间后重试 continue // 继续下一次重试 } conn = newConn continue // 重试读取 } 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) { global.GVA_LOG.Warn(fmt.Sprintf("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 // 重试写入 } 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) { global.GVA_LOG.Warn(fmt.Sprintf("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 // 重试写入 } 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 { _, err := conn.Write(frame) if err != nil { defer conn.Close() // 解析远程地址 addr, err := net.ResolveTCPAddr("tcp", conn.RemoteAddr().String()) if err != nil { // 处理错误... global.GVA_LOG.Error("解析错误 conn = " + addr.IP.String()) } model.ConnectionMap1.Delete(addr.IP.String()) return err } return nil } // 解析远程地址,提取 IP 地址和网络接口名称 func parseRemoteAddr(addr string) (string, string, error) { parts := strings.Split(addr, "%") if len(parts) != 2 { return "", "", fmt.Errorf("invalid remote address format: %s", addr) } return parts[0], parts[1], nil } func WriteAndReadDevice(frame []byte, conn net.Conn, former, after int) (data []byte, err error) { // 发送 Modbus RTU 帧 n, err := conn.Write(frame) if err != nil { global.GVA_LOG.Error("Error writing to connection:" + err.Error()) return } // 等待一段时间以接收响应 time.Sleep(1000 * time.Millisecond) // 读取响应 buffer := make([]byte, 1024) n, err = conn.Read(buffer) if err != nil { global.GVA_LOG.Error("Error reading from connection:" + err.Error()) return } // 检查读取的字节数是否足够 if n < former+after { err = fmt.Errorf("not enough bytes read to satisfy the slice range") return } // 返回子切片 return buffer[former : n-after], err }