Правильная техника потоковой очереди в C #?

Question

Я хотел реализовать службу Windows, которая собирает удаленные файлы с разделителями в папку для импорта в базу данных.Первоначально я предполагал, что FileSystemWatcher просматривает новые импортированные файлы и создает новый поток для импорта.

Я хотел знать, как правильно реализовать алгоритм для этого и какую технику использовать?Я иду в правильном направлении?

Answer 1

Традиционный подход заключается в создании конечного набора потоков (их может быть всего 1) и их отслеживании очереди блокировки.Код в обработчиках событий FileSystemWatcher ¹ ставит рабочие элементы в очередь, в то время как рабочие потоки удаляют их и обрабатывают.Это может выглядеть следующим образом, в котором используется класс BlockingCollection , доступный в .NET 4.0 или как часть загрузки Reactive Extensions .

Примечание:Код оставлен кратким и кратким для краткости.Вам придется расширять и укреплять его самостоятельно.

public class Example
{
  private BlockingCollection<string> m_Queue = new BlockingCollection<string>();

  public Example()
  {
    var thread = new Thread(Process);
    thread.IsBackground = true;
    thread.Start();
  }

  private void FileSystemWatcher_Event(object sender, EventArgs args)
  {
    string file = GetFilePathFromEventArgs(args);
    m_Queue.Add(file);
  }

  private void Process()
  {
    while (true)
    {
      string file = m_Queue.Take();
      // Process the file here.
    }
  }
}

Вы можете воспользоваться классом Task в TPL для более современного и ThreadPool -нравится подход.Вы бы начали новую задачу для каждого файла (или, возможно, пакетного), которые должны быть обработаны.Единственный недостаток, который я вижу при таком подходе, состоит в том, что будет сложнее контролировать количество соединений с базой данных, которые открываются одновременно.Это определенно не showtopper, и это не может быть проблемой.

¹ _{Известно, что FileSystemWatcher немного ненадежный, поэтому часто рекомендуется использовать вторичныйметод обнаружения изменений файла в случае, если они пропущены FileSystemWatcher.Ваш пробег может варьироваться в этом вопросе.}

Answer 2

Я разработал такой продукт для клиента.Служба отслеживала количество папок на наличие новых файлов, и когда файлы были обнаружены, файлы были прочитаны, обработаны (напечатаны на принтерах штрих-кодов), заархивированы и удалены.

Мы использовали обнаруженный слой «обнаружитель»файлы с использованием FileSystemWatcher или опроса в зависимости от среды (поскольку FileSystemWatcher не является надежным при мониторинге, например общих папок samba), слой «считыватель файлов» и слой «процессор».

Уровень «обнаружитель» обнаружил файлы и поместилимена файлов в списке, который обрабатывается слоем «читатель файла».Слой «обнаружитель» сигнализировал, что появились новые файлы для обработки, установив событие, которого ждал слой «считыватель файлов».

Затем слой «считыватель файлов» читает файлы (используя функцию повтора, поскольку вы можете получать уведомления о новых файлах до того, как файлы будут полностью записаны процессом, который создает файл).

После того, как слой «считыватель файлов» прочитал файл, был создан новый поток «процессор» с использованием ThreadPool.QueueWorkItem для обработки содержимого файла.

Когда файл был обработан,исходный файл был скопирован в архив и удален из исходного местоположения.Архив также регулярно очищался, чтобы не затопить сервер.Архив отлично подходит для поиска и устранения неисправностей.

В течение более двух лет он уже использовался в производстве в различных средах и оказался очень надежным.

Answer 3

Я выставил сервис, который тоже это делает.Я опрашиваю с помощью таймера, чей обработчик прошедших событий действует как супервизор, добавляя новые файлы в очередь и запуская настраиваемое количество потоков, которые потребляют очередь.Как только файлы обработаны, он перезапускает таймер.

Каждый поток, включая обработчик событий, перехватывает и сообщает обо всех исключениях.Служба всегда запущена, и я использую отдельное приложение пользовательского интерфейса, чтобы сообщить службе запустить и остановить таймер.Этот подход был непоколебим, и за несколько лет обработки сервис никогда не падал.

Answer 4

Создание темы для сообщения, скорее всего, будет слишком дорого. Если вы можете использовать .NET 4, вы можете запустить Task для каждого сообщения. Это будет запускать код в потоке пула потоков и, таким образом, сократить накладные расходы на создание потоков.

Вы также можете сделать нечто подобное с асинхронными делегатами , если .NET 4 не вариант. Однако в этом случае код становится немного сложнее. Это также использовало бы пул потоков и избавило бы вас от необходимости создавать новый поток для каждого сообщения.