Как удалить элемент из ConcurrentDictionary после окончательного завершения ContinueWith

Я не буду «принимать» этот ответ какое-то время, потому что мне не терпится увидеть, сможет ли кто-нибудь еще найти лучшее решение.

Коллега придумал решение, которое я немного подправил. Да, я знаю об иронии (?) использования оператора lock с оператором ConcurrentDictionary. У меня действительно нет времени прямо сейчас, чтобы посмотреть, будет ли лучший тип коллекции для использования. По сути, вместо того, чтобы просто делать ContinueWith() для существующих задач, мы заменяем задачу самой собой плюс еще одну задачу, прикрепляемую в конце, используя ContinueWith().

Какая разница? Рад, что вы спросили! :) Если бы мы только что выполнили ContinueWith(), то !worker.Value.IsCompleted вернуло бы true, как только первая задача в цепочке будет завершена. Однако, заменив задачу двумя (или более) связанными задачами, то, что касается коллекции, будет только одна задача, и !worker.Value.IsCompleted не вернет true, пока все задачи в цепочке завершены.

Признаюсь, я был немного обеспокоен заменой задачи самой собой+(новой задачей), потому что что, если задача окажется запущенной во время ее замены. Что ж, я тестировал живые дневные светы из этого и не столкнулся ни с какими проблемами. Я считаю, что происходит то, что, поскольку задача выполняется в своем собственном потоке, а коллекция просто содержит указатель на нее, работающая задача не затрагивается. Заменив его на саму + (новая задача), мы сохраняем указатель на исполняемый поток и получаем «уведомление», когда оно завершено, чтобы следующая задача могла «продолжиться», или IsCompleted возвращает true.

Кроме того, то, как работает цикл «очистки» и где он расположен, означает, что у нас будут «завершенные» задачи, висящие в коллекции, но только до следующего запуска «очистки», то есть в следующий раз, когда сообщение получено. Опять же, я провел много тестов, чтобы увидеть, не вызову ли я проблемы с памятью из-за этого, но моя служба никогда не использовала более 20 МБ ОЗУ, даже при обработке сотен сообщений в секунду. Нам пришлось бы получать довольно большие сообщения и иметь много длительных задач, чтобы это когда-либо вызывало проблему, но об этом следует помнить, поскольку ваша ситуация может отличаться.

Как и выше, в приведенном ниже коде nrtq = не относится к вопросу.

public interface IMessage
{
  long CompantId { get; set; }
  void Process();
}
public class CompanyMessage : IMessage
{ //implementation, nrtq }
public class ProductMessage : IMessage
{ //implementation, nrtq }

public class Controller
{
  private static ConcurrentDictionary<long, Task> _workers = new ConcurrentDictionary<long, Task>();
  //other needed declarations, nrtq

  public Controller(){//constructor stuff, nrtq }

  public StartSubscribers()
  {
    //other code, nrtq
    _companySubscriber.OnMessageReceived += HandleCompanyMsg;
    _productSubscriber.OnMessageReceived += HandleProductMsg;
  }

  private void HandleCompanyMsg(string msg)
  {
    //other code, nrtq
    QueueItUp(new CompanyMessage(message));
  }

  private void HandleProductMsg(string msg)
  {
    //other code, nrtq
    QueueItUp(new ProductMessage(message));
  }

  private static void QueueItUp(IMessage message)
  {
    lock(_workers)
    {
      foreach (var worker in Workers)
      {
        if (!worker.Value.IsCompleted) continue;
        Task task;
        Workers.TryRemove(worker.Key, out task);
      }
      var id = message.CompanyId;
      if (_workers.ContainsKey(id))
        _workers[id] = _workers[id].ContinueWith(x => message.Process());
      else
      {
        var task = new Task(y => message.Process(), id);
        _workers.TryAdd(id, task);
        task.Start();
      }
    }
  }