Обработка запросов из нескольких потоков в одном потоке - .NET Core

Question

Извините, если у этого вопроса уже есть ответ, но если это так, я не могу найти его на этом сайте.

Во-первых - этот вопрос является специфичным для .NET ядра (v1.1.0 на момент написания)

У меня есть сборка третьей стороны, которая будет обрабатывать только запросы, сделанные к нему, если они происходят в том же потоке , Эта сборка представляет собой библиотеку RabbitMQ и подробную информацию о проблеме, которая может быть или не быть релевантной can be found here. В принципе - у меня есть несколько потоков, которые потенциально могут вызвать эту сборку, но если запросы происходят из разных потоков, то это исключение.

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

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

я теперь (больно так) понимаю, что it is not possible in .NET core to begin-invoke an event in .Net core :(

Пример, который демонстрирует проблему:

public class Program 
{ 
    public static void Main(string[] args) 
    { 
     Program program = new Program(); 
     ManualResetEvent resetEvent = new ManualResetEvent(false); 
     program.StartNewThreadToBeCalled(resetEvent); 

     program.CallBobMultipleTimes(resetEvent); 

     Console.ReadLine(); 
    } 

    private void CallBobMultipleTimes(ManualResetEvent resetEvent) 
    { 
     resetEvent.WaitOne(); 
     for(int i=0 ; i<100 ; i++) 
      ThreadPool.QueueUserWorkItem(x=>CallBob(null, null)); //Can't BeginInvoke in .NET Core 
    } 

    private void StartNewThreadToBeCalled(ManualResetEvent resetEvent) 
    { 
     ThreadPool.QueueUserWorkItem(x=> 
     { 
      Bob bob = new Bob(); 

      CallBob += (obj, e)=> bob.InvokeMeOnOneThreadOnly(); 
      resetEvent.Set();     
      ManualResetEvent mre = new ManualResetEvent(false); 
      mre.WaitOne(); //Real implementation will block forever - this should be the only thread handles the processing of requests. 
     }); 
    } 

    public event EventHandler CallBob; 
} 

public class Bob 
{ 
    private List<int> ThreadIdentifiers = new List<int>(); 
    private static object SyncObject = new object(); 


    public void InvokeMeOnOneThreadOnly() 
    { 
     lock(SyncObject) 
     { 
      int currentThreadId = Thread.CurrentThread.ManagedThreadId; 
      if(ThreadIdentifiers.Any()) 
      { 
       if(!ThreadIdentifiers.Contains(currentThreadId)) 
        Console.WriteLine("Don't call me from multiple threads!"); 
      } 
      else 
       ThreadIdentifiers.Add(currentThreadId); 
     } 
    } 
} 

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

Есть ли лучший способ (чем тот, который я нашел) для меня обрабатывать несколько запросов из нескольких разных потоков в одном потоке в ядре .NET?

Answer 1

Сложно сказать, что вы пытаетесь сделать из кода. Но другой способ - использовать BlockingCollection<T>.Фоновый поток может добавить значения в эту коллекцию и один поток мог сидеть и попытаться получить значения из коллекции, например:

while(continueProcessingCollection) 
{ 
    if(blockingCollection.TryTake(out value, TimeSpan.FromSeconds(myTimeout))) 
    { 
     // process value, decide to toggle continueProcessingCollection 
    } 
} 

фона нить может добавить новые значения в коллекцию блокирующей:

blockingCollection.Add(newValue); 

И, где-то нужно создать переменную BlockingCollection:

var blockingCollection = new BlockingCollection<MyType>(); 

Answer 2

delegateBeginInvoke использует threadpool запланировать callback, который не решит вашу проблему. Это в основном проблема Producer–consumer с несколькими производителями и одним потребителем. .NET core содержит some structures реализация IProducerConsumerCollection<T> вы можете использовать для решения этой проблемы.

Не путайте с Control.BeginInvoke (winform) и Dispatcher.BeginInvoke (WPF), которые ставят очередь обратного вызова в потоке графического интерфейса.