使用C#Parallel.ForEach循环处理SFTP文件而不处理下载 [英] Processing SFTP files using C# Parallel.ForEach loop not processing downloads
问题描述
我正在使用Renci SSH.NET软件包2016版.我正在从外部服务器下载文件.通常，我每6秒钟可以下载大约一个文件，当您有成千上万个文件时，这很糟糕.我最近尝试将 foreach 循环更改为 Parallel.ForEach .这样做将文件下载时间更改为1.5秒.除了我检查文件时，它们都具有0 KB的文件，因此它没有下载任何内容.并行循环有什么问题吗?我是C#的新手，正在尝试缩短下载时间
I am using the Renci SSH.NET package version 2016. I am downloading files from an external server. I usually can download about one file every 6 seconds which is bad when you have thousands of files. I recently tried to change the foreach loops to Parallel.ForEach. Doing that changed the files downloaded times to 1.5 seconds. Except when I checked the files they all had 0 KB's so it did not download anything. Is there anything wrong with the parallel loop? I am new to C# and trying to improve download times
Parallel.ForEach(summary.RemoteFiles, (f, loopstate) =>
{
//Are we still connected? If not, reestablish a connection for up to a max of "MaxReconnectAttempts"
if (!sftp.IsConnected)
{
int maxAttempts = Convert.ToInt32(ConfigurationManager.AppSettings["MaxReconnectAttempts"]);
StatusUpdate(this, new Types.StatusUpdateEventArgs() { message = "SFTP Service has been connected from remote system, attempting to reconnect (" + sftpConnInfo.Host + ":" + sftpConnInfo.Port.ToString() + remotePath + " - Attempt 1 of " + maxAttempts.ToString() + ")", Location = locationName });
for (int attempts = 1; attempts <= maxAttempts; attempts++)
{
sftp.Connect();
if (sftp.IsConnected)
{
StatusUpdate(this, new Types.StatusUpdateEventArgs() { message = "SFTP Service - Connection reestablished (" + remotePath + ")", Location = locationName });
break;
}
else
{
if ((attempts + 1) <= maxAttempts)
{
StatusUpdate(this, new Types.StatusUpdateEventArgs() { message = "SFTP Service still disconnected from remote system, preparing another reconnect attempt (" + sftpConnInfo.Host + ":" + sftpConnInfo.Port.ToString() + remotePath + " - Attempt " + (attempts + 1).ToString() + " of " + maxAttempts.ToString() + ")", Location = locationName });
System.Threading.Thread.Sleep(2000);
}
else
{
//Max reconnect attempts reached - end the session and ensure the appropriate "failure" workflow is triggered
connectionLost = true;
}
}
}
}
if (connectionLost)
loopstate.Break();
// break;
totalFileCount++;
try
{
if (!System.IO.File.Exists(localSaveLocation + f.FileName))
{
System.Diagnostics.Debug.WriteLine("\tDownloading file " + totalFileCount.ToString() + "(" + f.FileName + ")");
System.IO.Stream localFile = System.IO.File.OpenWrite(localSaveLocation + f.FileName);
//Log remote file name, local file name, date/time start
start = DateTime.Now;
sftp.DownloadFile(f.FullName, localFile);
end = DateTime.Now;
//Log remote file name, local file name, date/time complete (increment the "successful" downloads by 1)
timeElapsed = end.Subtract(start);
runningSeconds += timeElapsed.TotalSeconds;
runningAvg = runningSeconds / Convert.ToDouble(totalFileCount);
estimatedSecondsRemaining = (summary.RemoteFiles.Count - totalFileCount) * runningAvg;
elapsedTimeString = timeElapsed.TotalSeconds.ToString("#.####") + " seconds";
System.Diagnostics.Debug.WriteLine("\tCompleted downloading file in " + elapsedTimeString + " " + "(" + f.FileName + ")");
downloadedFileCount++;
ProcessFileComplete(this, new Types.ProcessFileCompleteEventArgs() { downloadSuccessful = true, elapsedTime = timeElapsed.TotalSeconds, fileName = f.FileName, fullLocalPath = localSaveLocation + f.FileName, Location = locationName, FilesDownloaded = totalFileCount, FilesRemaining = (summary.RemoteFiles.Count - totalFileCount), AvgSecondsPerDownload = runningAvg, TotalSecondsElapsed = runningSeconds, EstimatedTimeRemaining = TimeSpan.FromSeconds(estimatedSecondsRemaining) });
f.FileDownloaded = true;
if (deleteAfterDownload)
sftp.DeleteFile(f.FullName);
}
else
{
System.Diagnostics.Debug.WriteLine("\tFile " + totalFileCount.ToString() + "(" + f.FileName + ") already exists locally");
downloadedFileCount++;
ProcessFileComplete(this, new Types.ProcessFileCompleteEventArgs() { downloadSuccessful = true, elapsedTime = 0, fileName = f.FileName + " (File already exists locally)", fullLocalPath = localSaveLocation + f.FileName, Location = locationName, FilesDownloaded = totalFileCount, FilesRemaining = (summary.RemoteFiles.Count - totalFileCount), AvgSecondsPerDownload = runningAvg, TotalSecondsElapsed = runningSeconds, EstimatedTimeRemaining = TimeSpan.FromSeconds(estimatedSecondsRemaining) });
f.FileDownloaded = true;
if (deleteAfterDownload)
sftp.DeleteFile(f.FullName);
}
}
catch (System.Exception ex)
{
// We log stuff here
}
});
推荐答案
我不知道为什么会得到空文件.尽管我怀疑您没有关闭 localFile 流.
I cannot tell why you get empty file. Though I'd suspect the fact that you do not close the localFile stream.
尽管，即使您的代码有效，如果您使用相同的下载连接，也几乎不会获得任何性能上的好处，因为SFTP传输往往受到网络延迟或CPU的限制.您必须使用多个连接来克服这一点.
Though, even if your code worked, you will get hardly any performance benefit if you use the same connection for the downloads, as SFTP transfers tend to be limited by a network latency or CPU. You have to use multiple connections to overcome that.
有关影响服务器SFTP传输速度的因素，请参见我的关于服务器故障的答案.
See my answer on Server Fault about factors that affect SFTP transfer speed.
实现一些连接池并每次选择一个空闲连接.
Implement some connection pool and pick a free connection each time.
简单的例子:
var clients = new ConcurrentBag<SftpClient>();
var opts = new ParallelOptions { MaxDegreeOfParallelism = maxConnections };
Parallel.ForEach(files, opts, (f, loopstate) => {
if (!clients.TryTake(out var client))
{
client = new SftpClient(hostName, userName, password);
client.Connect();
}
string localPath = Path.Combine(destPath, f.Name);
Console.WriteLine(
"Thread {0}, Connection {1}, File {2} => {3}",
Thread.CurrentThread.ManagedThreadId, client.GetHashCode(),
f.FullName, localPath);
using (var stream = File.Create(localPath))
{
client.DownloadFile(f.FullName, stream);
}
clients.Add(client);
});
Console.WriteLine("Closing {0} connections", clients.Count);
foreach (var client in clients)
{
client.Dispose();
}
另一种方法是启动固定数量的线程，每个线程具有一个连接，并让它们从队列中选择文件.
Another approach is to start a fixed number of threads with one connection for each and have them pick files from a queue.
有关实现的示例，请参阅我关于WinSCP .NET程序集的文章:
通过SFTP/FTP协议自动并行传输中的传输
For an example of implementation see my article for WinSCP .NET assembly:
Automating transfers in parallel connections over SFTP/FTP protocol
关于FTP的类似问题:
使用FluentFTP从FTP同时下载多个文件，并获得最大值
这篇关于使用C#Parallel.ForEach循环处理SFTP文件而不处理下载的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
