Golang中如何为连接添加持久性支持

要实现连接持久性，需要处理节点终止时的连接恢复机制。以下是关键改进点：

1. 添加连接重试逻辑：

func connectWithRetry(ha host.Host, peerid peer.ID, targetAddr ma.Multiaddr, maxRetries int) {
    for i := 0; i < maxRetries; i++ {
        ha.Peerstore().AddAddr(peerid, targetAddr, pstore.PermanentAddrTTL)
        
        s, err := ha.NewStream(context.Background(), peerid, "/p2p/1.0.0")
        if err == nil {
            go handleStream(s)
            return
        }
        
        log.Printf("连接失败，第%d次重试: %v", i+1, err)
        time.Sleep(time.Duration(i+1) * time.Second) // 指数退避
    }
    log.Println("达到最大重试次数，连接失败")
}

1. 修改主逻辑使用重试连接：

if *target != "" {
    ha.SetStreamHandler("/p2p/1.0.0", handleStream)
    
    ipfsaddr, err := ma.NewMultiaddr(*target)
    if err != nil {
        log.Fatalln(err)
    }
    
    pid, err := ipfsaddr.ValueForProtocol(ma.P_IPFS)
    if err != nil {
        log.Fatalln(err)
    }
    
    peerid, err := peer.IDB58Decode(pid)
    if err != nil {
        log.Fatalln(err)
    }
    
    targetPeerAddr, _ := ma.NewMultiaddr(
        fmt.Sprintf("/ipfs/%s", peer.IDB58Encode(peerid)))
    targetAddr := ipfsaddr.Decapsulate(targetPeerAddr)
    
    // 使用带重试的连接
    go connectWithRetry(ha, peerid, targetAddr, 5)
}

1. 增强流处理以检测断开：

func handleStream(s net.Stream) {
    log.Println("Got a new stream!")
    
    rw := bufio.NewReadWriter(bufio.NewReader(s), bufio.NewWriter(s))
    
    // 创建上下文用于控制goroutine
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    // 监控连接状态
    go func() {
        <-s.Closed()
        log.Println("连接关闭，准备重连")
        cancel()
        // 这里可以触发重连逻辑
    }()
    
    go readData(ctx, rw)
    go writeData(ctx, rw)
}

1. 修改读写函数支持上下文：

func readData(ctx context.Context, rw *bufio.ReadWriter) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
            str, err := rw.ReadString('\n')
            if err != nil {
                log.Printf("读取错误: %v", err)
                return
            }
            if str != "" {
                log.Printf("收到: %s", str)
            }
        }
    }
}

1. 添加心跳机制：

func startHeartbeat(rw *bufio.ReadWriter, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()
    
    for range ticker.C {
        _, err := rw.WriteString("ping\n")
        if err != nil {
            log.Println("心跳发送失败")
            return
        }
        rw.Flush()
    }
}

这些修改确保节点在连接断开时能够自动重连，同时通过心跳机制检测连接健康状态，实现真正的持久性连接。