Требуется отзыв о потоке кода

Question

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

 public class JobProcessor<TJob> : IJobProcessor<TJob> where TJob : class
 {
  private readonly IJobQueue<TJob> theJobQueue =
   new NullQueue<TJob>();

  private Thread processorThread;

  private bool shutdownRequested;

  private readonly IJobObserver<TJob> theObserver = new NullObserver<TJob>();

  public AutoResetEvent threadStartedEvent = new AutoResetEvent(false);

  private int processorThreadId = 0;

  private volatile TJob currentJob = null;

  public JobProcessor(IJobQueue<TJob> jobQueue, IJobObserver<TJob> observer)
  {
   if (observer != null)
   {
    theObserver = observer;
   }
   shutdownRequested = false;
   theJobQueue = jobQueue;
   CreateAndRunThread();
  }

  private void CreateAndRunThread()
  {
   processorThread = new Thread(Run)
                      {
                       Name = "Tpk Processor Thread", IsBackground = true
                      };
   processorThread.SetApartmentState(ApartmentState.STA);
   processorThread.Priority = ThreadPriority.BelowNormal;
   processorThread.Start();
  }

  public void Shutdown()
  {
   threadStartedEvent.WaitOne();

   shutdownRequested = true;

   theJobQueue.Interrupt(processorThreadId);
  }

  public TJob CurrentJob()
  {
   return currentJob;
  }

  private void Run()
  {
   processorThreadId = Thread.CurrentThread.ManagedThreadId;

   threadStartedEvent.Set();

   while (!BufferClearedAndShutDown())
   {
    try
    {
     ProcessNextMessage();
    }
    catch (ThreadAbortException)
    {
     CreateAndRunThread();
     break;
    }
    catch (Exception e)
    {  }
   }
  }

  private void ProcessNextMessage()
  {
   currentJob = theJobQueue.RetrieveJob();
   if (currentJob != null)
   {
    theObserver.ProcessMessage(this, currentJob);
   }
   currentJob = null;
  }

  private bool BufferClearedAndShutDown()
  {
   return theJobQueue.IsEmpty && shutdownRequested;
  }
 }
}

Answer 1

Это просто очередь производителя / потребителя? Есть много готовых примеров - я сам разместил один здесь несколько дней назад (но я не могу найти его в данный момент) ... скажу так - версия, которую я разместил, была много проще - то есть «явно нет ошибок», а не «нет очевидных ошибок». Я посмотрю, смогу ли я найти это ...

(редактировать: нашел его )

Дело в том; с этой версией рабочий поток просто делает:

T item;
while(queue.TryDequeue(out item)) {
    // process item
}
// queue has been closed and drained

Answer 2

Вы пытаетесь перехватить ThreadAbortException, а затем воссоздать себя? Я не уверен, что это возможно, но в любом случае это не очень хороший способ играть с ОС.

И вы потеряете работу, если произойдет исключение после currentJob = theJobQueue.RetrieveJob (); но перед Observer.ProcessMessage (this, currentJob);

И если ваше jobQueue не является поточно-ориентированным, вы должны добавить блокировку доступа к нему.

Answer 3

Очень сложно дать вам полезную обратную связь, не зная семантики IJobQueue, IJobObserver или IJobProcessor, но вот некоторые детали, которые выделяются:

1. processorThread, похоже, не нужен; может / должен быть просто локальным в CreateAndRunThread
2. shutdownRequested должен быть помечен как volatile, вызывать Thread.MemoryBarrier () после установки для shutdownReqeusted значения true или использовать Thread.VolatileWrite, чтобы установить поле shutdownRequested
3. зачем ждать запуска потока, прежде чем просить его отключить?
4. не знаю, зачем вам нужен threadStartedEvent, для чего он используется? особенно обнародовать это немного страшно
5. , не зная, как реализован IJobQueue.Interrupt, трудно сказать, есть ли там проблемы или нет
6. Кто использует CurrentJob и почему? чувствует себя рискованным
7. Хотя перехват вашей собственной ThreadAbortException будет перехватывать большинство случаев, он не будет перехватывать все. Вы можете использовать отдельный поток монитора, который вызывает Thread.Join, и после двойной проверки BufferClearedAndShutdown () вызывает CreateAndRunThread ()

Answer 4

Теперь, когда некоторые люди уже ответили, я чувствую, что можно добавлять свои собственные комментарии, никого не трогая: -)

1. Почему вы хотите дождаться начала потока, прежде чем его отключить?
2. Запись идентификатора потока и использование его для определения того, какой поток может его закрыть?
3. currentJob помечен как энергозависимый, но не shutdownRequested (и он не установлен в пределах блокировки)
4. Зачем смешивать очередь + рабочий в одной? Почему бы не выделить отдельную очередь, а затем работать с ней один или несколько потоков.
5. Не могу понять, почему он ожидал исключение ThreadAbortException, чтобы отбросить текущий рабочий поток и создать новый.
6. Несмотря на комментарий в верхней части кода о том, как можно просто расширить класс для использования нескольких рабочих потоков, я не думаю, что этот дизайн вообще упрощает переход к нескольким рабочим потокам (в отличие от подхода, когда очередь / работники отделены).

Я не многопоточный гуру или что-то в этом роде, поэтому я хотел опубликовать его, просто чтобы убедиться, что не только я думал, что этот код выглядит немного хитроумным.

Answer 5

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

в пункте 2 Рэнди, я бы просто использовал оператор блокировки, а не барьер памяти, это рекомендуемый MSDN подход, если вы не используете несколько процессоров Itanium на своих компьютерах (со слабым упорядочением памяти)

http://msdn.microsoft.com/en-us/library/system.threading.thread.memorybarrier.aspx

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