从Blob存储向SFTP服务器批量上传文件超时及效率问题排查

问题背景 #

• 从Blob存储批量上传文件到SFTP服务器时，频繁抛出Renci.SshNet.Common.SshConnectionException，堆栈信息如下：

  at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw (System.Private.CoreLib, Version=8.0.0.0, Culture=neutral, PublicKeyToken=7cec85d7bea7798e)
  at Renci.SshNet.Session.WaitOnHandle (Renci.SshNet, Version=2024.0.0.0, Culture=neutral, PublicKeyToken=1cee9f8bde3db106)
  

• 每次上传后已关闭SFTP连接，但仍出现超时；60个小文件上传耗时约5分钟，效率极低。
• 当前执行流程：
  1. 按前缀从Blob获取文件列表
  2. 逐个下载文件到内存流
  3. 每次上传新建SFTP连接并完成上传

当前代码实现 #

public async Task<bool> DownloadListOfFiles(ILogger log, string prefix, int? segmentSize)
{
    bool isSuccess = true;
    try
    {
        var resultSegment = containerClient.GetBlobsByHierarchyAsync(prefix: prefix, delimiter: "/")
            .AsPages(default, segmentSize);

        await foreach (Page<BlobHierarchyItem> blobPage in resultSegment)
        {
            foreach (BlobHierarchyItem blobhierarchyItem in blobPage.Values)
            {
                if (blobhierarchyItem.IsPrefix)
                {
                    await DownloadListOfFiles(log, blobhierarchyItem.Prefix, null);
                }
                else
                {
                    var blobClient = containerClient.GetBlobClient(blobhierarchyItem.Blob.Name);
                    string FileName = blobhierarchyItem.Blob.Name;
                    var remoteFilePath = $"SFTPPath/ {FileName}";
                    using (var stream = new MemoryStream())
                    {
                        await blobClient.DownloadToAsync(stream);
                        stream.Position = 0;
                        await PostToSFTP(stream.ToArray(), log, remoteFilePath );
                    }
                }
            };
        }
    }
    catch (RequestFailedException e)
    {
        isSuccess = false;
    }
    return isSuccess;
}

private async Task PostToSFTP(byte[] fileBytes, ILogger log, string remoteFilePath)
{
    bool isSuccess = true;
    using (SftpClient sftpclient = new SftpClient(SFTPHost, SFTPPort, SFTPUserName, SFTPPassword))
    {
        try
        {
            sftpclient.Connect();
            sftpclient.KeepAliveInterval = new TimeSpan(0, 1, 0);
            sftpclient.OperationTimeout = new TimeSpan(0, 1, 0);

            using (Stream stream = new MemoryStream(fileBytes))
            {
                sftpclient.UploadFile(stream, remoteFilePath, true);
            }
            sftpclient.Disconnect();
        }
        catch (SshException ex)
        {
            sftpclient.Disconnect();
            isSuccess = false;
            if (ex.InnerException != null && ex.InnerException.Message.Contains("Socket read operation has timed out"))
            {
                Thread.Sleep(100);
            }
        }
        catch (Exception ex)
        {
            sftpclient.Disconnect();
            isSuccess = false;
        }
        finally
        {
            sftpclient.Disconnect();
            sftpclient.Dispose();
        }
    }
}

优化方案 #

1. 复用SFTP连接，避免频繁创建销毁 #

每次上传新建连接会产生大量握手开销，且频繁连接易触发服务器端超时限制。改为复用单个连接处理所有文件：

• 在遍历文件前建立一次SFTP连接，上传完成后统一关闭
• 增加连接存活检查，断开时自动重连

修改后的核心逻辑示例：

public async Task<bool> DownloadListOfFiles(ILogger log, string prefix, int? segmentSize)
{
    bool isSuccess = true;
    SftpClient sftpclient = null;
    try
    {
        // 初始化一次SFTP连接
        sftpclient = new SftpClient(SFTPHost, SFTPPort, SFTPUserName, SFTPPassword);
        sftpclient.Connect();
        sftpclient.KeepAliveInterval = new TimeSpan(0, 0, 30); // 缩短保活间隔，避免闲置超时
        sftpclient.OperationTimeout = new TimeSpan(0, 2, 0); // 适当延长操作超时

        var resultSegment = containerClient.GetBlobsByHierarchyAsync(prefix: prefix, delimiter: "/")
            .AsPages(default, segmentSize);

        await foreach (Page<BlobHierarchyItem> blobPage in resultSegment)
        {
            foreach (BlobHierarchyItem blobhierarchyItem in blobPage.Values)
            {
                if (blobhierarchyItem.IsPrefix)
                {
                    await DownloadListOfFiles(log, blobhierarchyItem.Prefix, null);
                }
                else
                {
                    var blobClient = containerClient.GetBlobClient(blobhierarchyItem.Blob.Name);
                    string fileName = blobhierarchyItem.Blob.Name;
                    var remoteFilePath = $"SFTPPath/{fileName}"; // 移除路径空格，避免路径错误
                    using (var stream = new MemoryStream())
                    {
                        await blobClient.DownloadToAsync(stream);
                        stream.Position = 0;
                        // 复用已有的SFTP连接上传
                        await UploadToSFTPWithReuse(sftpclient, stream, remoteFilePath, log);
                    }
                }
            };
        }
    }
    catch (RequestFailedException e)
    {
        isSuccess = false;
        log.LogError(e, "Blob操作失败");
    }
    catch (SshException e)
    {
        isSuccess = false;
        log.LogError(e, "SFTP连接或操作失败");
    }
    finally
    {
        sftpclient?.Disconnect();
        sftpclient?.Dispose();
    }
    return isSuccess;
}

private async Task UploadToSFTPWithReuse(SftpClient sftpclient, Stream fileStream, string remoteFilePath, ILogger log)
{
    try
    {
        if (!sftpclient.IsConnected)
        {
            sftpclient.Connect();
        }
        sftpclient.UploadFile(fileStream, remoteFilePath, true);
    }
    catch (SshException ex)
    {
        log.LogError(ex, $"上传文件 {remoteFilePath} 失败");
        // 尝试重连一次
        if (!sftpclient.IsConnected)
        {
            sftpclient.Connect();
            fileStream.Position = 0;
            sftpclient.UploadFile(fileStream, remoteFilePath, true);
        }
        else
        {
            throw;
        }
    }
}

2. 优化内存与流处理 #

• 避免将文件转为byte[]，直接传递MemoryStream减少内存拷贝
• 替换Thread.Sleep为异步延迟，避免阻塞线程

3. 增加指数退避重试机制 #

针对Socket超时异常，实现重试逻辑提升可靠性：

private async Task UploadWithRetry(SftpClient sftpclient, Stream fileStream, string remoteFilePath, ILogger log, int maxRetries = 3)
{
    int retryCount = 0;
    while (retryCount < maxRetries)
    {
        try
        {
            if (!sftpclient.IsConnected)
            {
                sftpclient.Connect();
            }
            fileStream.Position = 0;
            sftpclient.UploadFile(fileStream, remoteFilePath, true);
            return;
        }
        catch (SshException ex) when (ex.InnerException?.Message.Contains("Socket read operation has timed out") == true)
        {
            retryCount++;
            log.LogWarning($"上传超时，重试 {retryCount}/{maxRetries}");
            await Task.Delay(TimeSpan.FromSeconds(Math.Pow(2, retryCount))); // 指数退避延迟
        }
    }
    throw new Exception($"上传文件 {remoteFilePath} 多次重试后失败");
}

4. 有限度并行上传（可选） #

如果SFTP服务器支持多连接，可使用Parallel.ForEachAsync实现并行上传，设置合理并行度（如4-8），避免连接过多被服务器拒绝。

5. 调整超时参数 #

• 缩短KeepAliveInterval至30秒，确保闲置时发送保活包
• 适当延长OperationTimeout，避免网络波动触发不必要的超时

内容的提问来源于stack exchange，提问作者Rajib Mahata