Как обрабатывать исключения в TPL Dataflow - Producer/Consumer

Question

У меня есть следующее тестовое приложение, которое имитирует мой сценарий приложения:

class Program 
{ 
    static void Main() 
    { 
     Console.WriteLine("***Press R = record to new file, S = Stop Recording, E = Exit"); 
     var timer = new Timer(Callback, null, 0, 1000); 
     while(!_Close) HandleInput(Console.ReadLine()); 
     Console.WriteLine("Finished"); 
    } 

    private static bool _Close; 

    private static void HandleInput(string input) 
    { 
     switch (input.ToLower()) 
     { 
      case "r": CreateWriter(); 
       break; 
      case "s": Console.WriteLine("File Closed: {0}", _FileWriter.Name); 
       _FileWriter.Dispose(); 
       _FileWriter = null; 
       break; 
      case "e": 
       _Close = true; 
       break; 
     } 
    } 

    private static void CreateWriter() 
    { 
     if (_FileWriter != null) 
      _FileWriter.Dispose(); 
     string filename = Path.Combine("C:\\", string.Format("{0:yyyy_MM_dd HH_mm_ss}.txt",DateTime.Now)); 
     _FileWriter = new AsyncFileWriter(filename); 
     Console.WriteLine("New File Created: {0}", filename); 
    } 

    private static void Callback(object state) 
    { 
     if (_FileWriter != null) 
      _FileWriter.Produce(MakeData()); 
    } 

    private static byte[] MakeData() 
    { 
     string data = string.Empty; 
     for (int i = 0; i < 50; i++) 
     { 
      data += string.Format("{0:yyyy-MM-dd HH:mm:ss.fff}{1}", DateTime.Now, Environment.NewLine); 
     } 
     return Encoding.UTF8.GetBytes(data); 
    } 

    private static AsyncFileWriter _FileWriter; 
} 

Он использует следующий класс в качестве потребителей (/ продюсер):

public class AsyncFileWriter : IDisposable 
{ 
    private readonly FileStream _Filewriter; 
    private readonly Task _WriteTask; 
    private readonly BufferBlock<byte[]> _BufferBlock; 

    public AsyncFileWriter(string filename) 
    { 
     _Filewriter = new FileStream(filename, FileMode.CreateNew, FileAccess.Write, FileShare.None); 
     _BufferBlock = new BufferBlock<byte[]>(); 
     _WriteTask = WriteToFile(); 
    } 

    public void Produce(byte[] data) 
    { 
     _BufferBlock.Post(data); 
    } 

    public long Filesize { get; private set; } 

    public string Name { get { return _Filewriter.Name; } } 

    private async Task WriteToFile() 
    { 
     while (await _BufferBlock.OutputAvailableAsync()) 
     { 
      byte[] data = _BufferBlock.Receive(); 
      await _Filewriter.WriteAsync(data, 0, data.Length); 
      Filesize = _Filewriter.Length; 
     } 
    } 

    private async Task Complete() 
    { 
     _BufferBlock.Complete(); 
     await Task.WhenAll(_WriteTask, _BufferBlock.Completion); 
     //now close the file 
     _Filewriter.Dispose(); 
    } 

    public void Dispose() 
    { 
     Complete(); 
    } 
} 

важные вещи, чтобы отметить следующие:

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

У меня есть несколько вопросов, так как я совершенно новичок в библиотеке потоков данных TPL.

1. Во-первых, является ли моя реализация звуком продюсера/потребителя (то есть AsyncFileWriter)?
2. Является ли моя реализация IDisposable ОК?
3. Как обрабатывать исключения в Потребителе (то есть WriteToFile)? Мне действительно нужно получить уведомление, если что-то пойдет не так, но, как вы видите, единственный публичный метод - Produce(), и исключение не произойдет здесь ... Существует ли наилучшая практика для этого шаблона?

Answer 1

Dispose является своего рода сломанным, потому что он не ждет, пока вся деятельность не будет закрыта. Это, вероятно, не то, что вы хотите. См. http://blog.stephencleary.com/2013/03/async-oop-6-disposal.html.

AsyncFileWriter имеет неограниченный буфер. Если файловый писатель перегружен, это будет потреблять все больше и больше памяти.

AsyncFileWriter кажется ОК в противном случае.

Но поскольку файловая система выполняет свою собственную буферизацию, вам, возможно, удастся избежать использования ничего подобного AsyncFileWriter. Просто запись в файл может быть достаточно хорошей.

относительно исключений: для этого вы можете указать событие. Или, в случае исключения, вы закрываете письмо, помните об исключении и позволяете потребителям класса изучать исключение. Кроме того, продукт должен начинать с сбоя. Dispose не должен терпеть неудачу, но перед тем, как избавиться от пользователей класса, необходимо закрыть его и проверить исключения.

Так что либо используйте событие, либо переместите класс в состояние ошибки.

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