Puli wątków "Naprawiono"/"Równoważenie obciążenia" C#?

Question

Mam składnik innej firmy, który jest "drogi" do zakręcenia. Ten komponent nie jest bezpieczny dla wątków. Ten składnik jest hostowany wewnątrz usługi WCF (na razie), więc ... za każdym razem, gdy połączenie przychodzi do usługi, muszę dokonać aktualizacji komponentu.

Zamiast tego chciałbym mieć pulę powiedzmy 16 wątków, z których każdy rozpędziłby swoją kopię komponentu i miał mechanizm wywoływania metody i rozdzielił go na jeden z 16 wątków. zwrócona wartość.

Więc coś prostego jak:

var response = threadPool.CallMethod(param1, param2); 

Jego porządku na wezwanie do blokowania aż dostaje odpowiedź jak ja potrzebuję odpowiedzi, aby kontynuować.

Wszelkie sugestie? Być może przesadzam i ConcurrentQueue, który jest obsługiwany przez 16 wątków wykona zadanie, ale teraz pewny, w jaki sposób zwracana wartość metody zostanie zwrócona do osoby dzwoniącej?

Answer 1

WCF będzie już używać puli wątków do zarządzania swoimi zasobami, więc jeśli dodasz warstwę zarządzania wątkami, to będzie to tylko źle. Unikaj tego, jeśli to możliwe, ponieważ będziesz mieć spór o swoje zgłoszenia serwisowe.

W tej sytuacji powinienem użyć pojedynczego pojedynczegoThreadLocal lub wątku statycznego, który zostanie zainicjalizowany z twoim drogim obiektem. Następnie będzie dostępny dla wątku puli wątków.

Zakłada to, że twój obiekt jest w porządku na wątku MTA; Zgaduję, że pochodzi to z twojego wpisu, ponieważ wygląda na to, że rzeczy działają, ale są wolne.

Istnieje obawa, że ​​powstaje zbyt wiele obiektów i zużywa się zbyt dużo pamięci, ponieważ pula jest zbyt duża. Zobacz jednak, czy tak jest w praktyce, zanim zrobisz cokolwiek innego. Jest to bardzo prosta strategia wdrożenia tak łatwa do przeprowadzenia. Stawaj się bardziej złożony, jeśli naprawdę potrzebujesz.

Answer 2

Przede wszystkim, zgadzam się z @briantyler: ThreadLocal<T> lub wątek pola statyczne jest prawdopodobnie to, co chcesz. Powinieneś pójść z tym jako punktem wyjścia i rozważyć inne opcje, jeśli nie spełnia Twoich potrzeb.

Skomplikowana, ale elastyczna alternatywa to pojedyncza pula obiektów. W swojej najprostszej postaci twój typ basen będzie wyglądać następująco:

public sealed class ObjectPool<T> 
{ 
    private readonly ConcurrentQueue<T> __objects = new ConcurrentQueue<T>(); 
    private readonly Func<T> __factory; 

    public ObjectPool(Func<T> factory) 
    { 
     __factory = factory; 
    } 

    public T Get() 
    { 
     T obj; 
     return __objects.TryDequeue(out obj) ? obj : __factory(); 
    } 

    public void Return(T obj) 
    { 
     __objects.Enqueue(obj); 
    } 
} 

Nie wydaje strasznie przydatne, jeśli myślisz o rodzaju T w warunkach prymitywnych klas lub kodowanym (tj ObjectPool<MyComponent>), w basenie nie ma wbudowanych żadnych elementów sterujących wątkami. Możesz jednak zastąpić monotonę typu T dla monady i uzyskać dokładnie to, co chcesz.

Basen inicjalizacji:

Func<Task<MyComponent>> factory =() => Task.Run(() => new MyComponent()); 
ObjectPool<Task<MyComponent>> pool = new ObjectPool<Task<MyComponent>>(factory); 

// "Pre-warm up" the pool with 16 concurrent tasks. 
// This starts the tasks on the thread pool and 
// returns immediately without blocking. 
for (int i = 0; i < 16; i++) { 
    pool.Return(pool.Get()); 
} 

Zastosowanie:

// Get a pooled task or create a new one. The task may 
// have already completed, in which case Result will 
// be available immediately. If the task is still 
// in flight, accessing its Result will block. 
Task<MyComponent> task = pool.Get(); 

try 
{ 
    MyComponent component = task.Result; // Alternatively you can "await task" 

    // Do something with component. 
} 
finally 
{ 
    pool.Return(task); 
} 

Ta metoda jest bardziej skomplikowane niż utrzymanie składnika w ThreadLocal lub nitki statycznego pola, ale jeśli trzeba coś zrobić fantazyjny jak ograniczenie liczba połączonych instancji, abstrakcja puli może być całkiem przydatna.

EDIT

Podstawowe „stałe ustawiony instancji X” realizacja basen z Get który blokuje raz na basen został osuszony:

public sealed class ObjectPool<T> 
{ 
    private readonly Queue<T> __objects; 

    public ObjectPool(IEnumerable<T> items) 
    { 
     __objects = new Queue<T>(items); 
    } 

    public T Get() 
    { 
     lock (__objects) 
     { 
      while (__objects.Count == 0) { 
       Monitor.Wait(__objects); 
      } 

      return __objects.Dequeue(); 
     } 
    } 

    public void Return(T obj) 
    { 
     lock (__objects) 
     { 
      __objects.Enqueue(obj); 

      Monitor.Pulse(__objects); 
     } 
    } 
}