3

好的，情况就是这样……我有一个每秒生成大约 8 个文件的应用程序。每个文件为 19-24kb。这会每分钟生成大约 10 到 11 MB。这个问题不是关于如何 ftp，因为我已经有了那个解决方案......问题更多的是关于如何跟上数据流（在大多数情况下只有 2mb 的上传带宽，除非我正在前往客户端站点有一个大管道）。我不在乎 ftp 是否需要更长的时间来传输流量，但我想知道是否有人知道如何批处理文件以移动它们，以便在 ftp 进程完成时它只会删除它传输的那些文件然后继续下一批。这是我的想法：

应用程序多线程，第一个线程运行应用程序，第二个线程是一个计时器，它每“N”分钟创建一个文本文件，其中包含在该时间跨度内创建的所有文件。StreamRead 文件并将文本中的文件移动到另一个位置（可能创建一个临时文件夹），然后 ftp 这些文件，然后删除文件、文件夹和文本文件......同时，正在写入更多文本文件和临时文件正在创建的文件夹。这听起来可行吗？我会接受任何人的建议，只是寻找最快和最可靠的路径。

请不要要求查看代码，考虑到我们正在处理假设，没有理由看到它。

4

7 回答 7

4

我会创建一个服务，并使用 FileSystemWatcher、System.Threading.Timer 或两者将传入的文件添加到并发集合中（如果 FileSystemWatcher 的缓冲区溢出，它可能会丢失文件，所以最好有一个计时器来接收任何文件错过了）。当文件进入时，我会将它们移动到一个单独的文件夹中，并使用 .NET 4.0 任务处理它们。然后，我将在原始任务的后续步骤中进行任何必要的后期处理。您可以拥有处理任何故障的延续步骤和成功时发生的不同延续步骤。这些任务中的每一个都将在线程池中启动一个线程并为您管理。

这是来自http://msdn.microsoft.com/en-us/library/dd997415.aspx的 OnlyOnFaulted 延续任务的示例。您可以有第二个延续任务，只有在成功时才会运行。

var task1 = Task.Factory.StartNew(() =>
{
    throw new MyCustomException("Task1 faulted.");
})
.ContinueWith((t) =>
    {
        Console.WriteLine("I have observed a {0}",
            t.Exception.InnerException.GetType().Name);
    },
    TaskContinuationOptions.OnlyOnFaulted);
于 2011-05-27T14:35:50.760 回答
1

如果不真正了解为什么需要将所有工作保留在单个应用程序中并处理线程复杂性的更多细节，人们可能会争辩将生成文件的部分和 FTP 文件的部分保留在单独的应用程序中。

责任分离。确保每个应用程序只完成一项工作，并且正确且快速地完成。

一个服务或应用程序（桌面/网络）生成文件。

另一个监视文件夹并将任何传入文件移动到临时文件的服务，执行它需要做的事情，FTP 和删除。

看到我不知道您的设置以及您从哪里获取文件内容，将其写入单个应用程序可能是您所建议的最佳选择。

基本上回答你的问题。是的，你想做的事情听起来确实可行。您如何实施它以及您对实施的满意程度取决于您。

如果您在实施过程中遇到问题，请随时在新威胁中发布任何问题，并附上一些代码示例，说明您如何实施特定功能以及遇到的问题。

在那之前，假设您认为能够管理您需要实现的目标的任何方法都是完全有效的。

编辑

看到您说您已经完成了生成文件的应用程序，并且您已经有了一个解决方案，FTP 意味着使用 2 个单独的应用程序听起来更合理。

然后，您所需要的只是围绕 FTP 解决方案和快乐的日子包装服务。如果它已经在工作，则无需干扰生成文件的原始应用程序。

为什么要冒险破坏它，除非您必须在其中添加 ftP 功能而您别无选择。

于 2011-05-27T14:36:58.890 回答
1

我在以前的工作中从事过类似的工作。我将外部进程转储文件放在某个文件夹上。这是我遵循的算法：

  1. 在转储文件的源目录上运行 FileSystemWatcher
  2. 找到新文件后，按日期升序处理目录中的所有文件。（在你的情况下 ftp 文件）
  3. 处理文件后，我将它们移动到 Processed 目录（在您的情况下，您可以删除它们）

需要考虑的事项：

  1. 我可以拥有多少个打开的 ftp 连接/处理线程
  2. FileSystemWatcher 可以并且将在处理另一个文件时引发事件。如何处理它/将它发送到适当的线程
于 2011-05-27T14:52:42.897 回答
0

您需要在文件的生产者和消费者（FTP 主机）之间插入一个队列，以便在生产者速度过快时能够缓冲文件。这需要某种形式的多线程甚至多个进程。

您提出了一个解决方案，其中队列是文件系统，这很有可能，但在许多情况下并不理想。您必须正确锁定以避免传输半满或空文件等。如果您决定使用文件系统，我的经验FileSystemWatcher不能用于此目的。使用计时器来运行任务说每秒拾取新文件要可靠得多。

其他队列技术可能是内存队列（但您必须考虑如何处理崩溃）、私有 Microsoft 消息队列或 SQL Server 代理队列。最佳解决方案很大程度上取决于您的要求。

FTP 不是真正事务性的，您可能决定使用非事务性队列（MSMQ 和 SQL Server Broker 都是事务性的），但您仍应尝试围绕创建文件的事务的概念构建应用程序，排队并交付。如果无法交付，则将其留在队列中，稍后重试交付。如果它不能被排队，生产者应该重试排队等等。你不想要一个文件永远不会被传递或被传递两次的情况。

从您的问题中不清楚您将如何使用 FTP，但我建议您使用开源或商业库来直接从您的应用程序中使用 FTP，而不是使用ftp.exe. 这将允许您的应用程序智能地保持 FTP 连接打开以避免过度重新连接等。

您还应该考虑如何处理队列增长过大的情况。一种选择可能是停止生产者，直到队列大小减少到阈值以下。

于 2011-05-27T14:57:21.850 回答
0

作为在这种情况下的 FTP 服务器所有者，我还要求您找到一种尽可能保持登录状态的方法。

登录/注销通常比单个文件传输更“昂贵”（在计算、配置阻塞等方面）。

于 2011-06-21T03:29:20.077 回答
0

我会使用 BlockingCollections 建立一个线程链。

一个生产者线程使用计时器或 FileSystemWatcher 等读取可用文件，并将它们存储在 BlockingCollection 中。它还将文件存储在一个列表中，以确保它们只添加一次。 

var availableFiles = new BlockingCollection<string>();
var processedFiles = new BlockingCollection<string>();
var newFiles = new HashSet<string>();

...
lock (newFiles) {
    foreach (var file in Directory.GetFiles())
        if (!newFiles.Contains(file)) {
            availableFiles.Add(file);
            newFiles.Add(file);
        }
}

一个或多个 ftp 线程发送文件，然后将它们放入已处理的集合中

foreach (var file in availableFiles.GetConsumingEnumerable()) {
   SendFileOverFtp(file);
   processedFiles.Add(file);
}

一个线程清理处理过的文件

foreach (var file in processedFiles.GetConsumingEnumerable()) {
    lock (newFiles) {
       File.Delete(file);
       newFiles.Remove(file);
    }
}

另一种选择是让生产线程也将文件读入内存并删除它们。在这种情况下，您可以跳过最后一个阶段和 newFiles 集合

于 2011-05-28T19:18:51.860 回答
0
  1. 启动一个每秒触发一次的计时器。
  2. 在计时器的经过事件处理程序中，停止计时器。
  3. 获取传入目录中所有文件的列表。
  4. 尝试以独占方式打开每个文件。这可以防止您读取仍在写入的文件。
  5. 将每个文件复制到暂存目录并将其从传入目录中删除。
  6. 移动列表中的所有文件后，通过 FTP 发送暂存目录中的文件。
  7. 将文件通过 FTP 传输后，将它们从暂存目录中删除。
  8. 启动计时器。

计时器的经过处理程序在线程池上为您运行，您应该需要任何更高级的线程管理。由于您的主要限制是您的 FTP 带宽，因此在上传文件之前与其他线程进行任何其他操作几乎没有优势。

这种方法可以在系统崩溃时为您提供保护。暂存目录中未发送的文件将在下一个周期中提取。传入目录中的文件也是如此。

如果您的 FTP 接收端可以处理压缩文件，您将通过压缩暂存目录的内容并将其作为一个文件发送来提高吞吐量。

于 2011-05-27T19:06:03.200 回答