ConcurrentDictionary, ConcurrentBag, ConcurrentQueue ve ConcurrentStack ile Thread Safe Collectionlar

Selamlar,

Bir konuyu anlamanın en iyi yolu, ona ihtiyaç duymaktır. İcatlar, ihtiyaçlardan doğar ilkesinden yola çıkarak, gelin öncelikle karşımıza çıkabilecek bir sorunu hep beraber tartışalım ve sonrada ona çözüm yollarını arayalım.

Öncelikle, kuyruktaki elemanları sıralayan sıradan bir console uygulaması yazalım.

Şimdi bu kuyruktaki elemanları 2 farklı  asenkron methoddan sıralatmak istersek, aşağıdaki “Queue empty” hatasını alırız. Çünkü asenkron methodlar birbirini beklememekte ve bir thread içinde “while (counter.Count > 0)” koşulu içine girildikten sonra, başka bir thread tarafından bozulmaktadır. Ve boş bir queue içinden eleman alınılmaya çalışılmaktadır. İlgili kodu aşağıda görebilirsiniz.

Hikaye: Bu tarz durumlar, iş hayatında bolca karşımıza çıkabilmektedir. Ve çözüm yolu ararken bir okadar da can sıkıcı olabilmektedir. Mesela sıralı ID’li bir faturalama işlemi yapıyorsunuz. İlgili bir collection’dan önceden rezerve edilen 20 tekil(unique) ID’yi, yazdırılan faturalara sıra ile vericeksiniz. Ya da sırada bekleyen 10’larca client var. Bir sohbet ya da Okey odası yapıyorsunuz. Yer açılan odaya sıradaki kişiyi asenkron olarak atayacaksınız. Kısaca örnekler daha da çoğaltılabilir.

Bu aşamadan sonra “System.Collections.Concurrent” namespace’inden türüyen thread-safe collectionları hep beraber inceleyeceğiz.

Çözüm: 

ConcurrentQueue<T>:

Asenkron yapılarda “Queue“, yukarıdaki örnekte de görüldüğü gibi kullanılabilecek doğru bir araç değildir. Bunun için “ConcurrentQueue” collection hiçte yanlış bir seçim olmaz. Böylece her iki thread’in birbirini ezmemesi, “TryDequeue()” methodu ile kontrol edilerek sağlanabilir. Aşağıda görüldüğü gibi, sıralamanın sırası değişebilmektedir. Önce 2 sonra 1 çalışabilmektedir. Bu asenkron programlamanın doğasında vardır. Ama esas önemli olan kısım, hiç bir thread diğerinin işine elini sokmamakta ve tüm dizi elemanları sadece 1 kere farklı tasklar tarafından çekilerek ekrana basılmaktadır.

Not:TryDequeue()” ==> yerine “TryPeek()” methodu kullanılsa idi, kayıtlar sadece okunup silinmeyeceği için, while methodu “while (counter.Count > 0)” sonsuza kadar çalışacaktı.

ConcurrentDictionary: Diyelim ki bir Dictionary’ye, 2 farklı async Task methoddan aynı anda [ 10200 – 10230 ] arası ID değerleri key olarak, Task adları string value olarak doldurulsun. Burada Queue’den farklı olarak istenen ID’ye göre karşılık gelen value değeri yani atandığı Task adı bulunacaktır. Öncelikle gelin kodu, normal “Dictionary” ile yazalım, sonra sonuçlarını ve çözüm yolunu tartışalım.
  1. “AddDict1()”: async methodu ” for (int i = 10200; i < 10230; i++)” ile [ 10200-10230 ] arasındaki sayıları dictionary’e eklemek için kullanılmıştır.
  2. “if (!dict.ContainsKey(i))” :  Aynı ID’nin yani key’in, list’e eklenmemesi için var mı yok mu diye bakılır.
  3. “dict.Add(i, “AddDict1″);” : Aynı ID değeri yok ise, dictionary list’e ekleme işlemi yapılır.
  4. “AddDict2()”: Methodu da [ 10200-10230 ] arasındaki sayıları, aynı dictionary list’e eklemeye çalışmaktadır.
  5. “AddKeyList()”: İlgili methodda “AddDict1() ve AddDict2()” methodları task1 ve task2 olarak atanır ==>” var task1 = AddDict1(); var task2 = AddDict2();”
  6. “await Task.WhenAll(task1, task2)” : Tüm asenkron methodların tamamlanması beklenir.

Sonuç: Aşağıdaki sonuca bakıldığında 1. Thread içinde ” !dict.ContainsKey(i)” koşulu geçilip, 2.Thread’de ilgili elemanın eklendikten sonra aynı elemanın 1.Thread’de de eklendiği görülmektedir. Demeki Threadler için “Dictionary” yerine Thread Safe olan ConcurrentDictionary’nin kullanılması daha doğrudur.

Şimdi gelin aynı örneği ConcurrentDictionary ile yapalım. Aşağıda görüldüğü gibi “dict = new ConcurrentDictionary<int, string>()” ile farklı threadlerin birbirine karışması, birbirini ezmesi engellenmiştir. “dict.TryAdd()” methodu ile aynı key’lerin eklenmesine izin verilmemiştir.

Sonuç: Aşağıda görüldüğü gibi farklı threadler ile dolan ConcurrentDictionary, kesinlikle aynı item’ı bünyesine almamış ve threadler arasında senkronizasyonu sağlamıştır.

ConcurrentDictionary’nin : Aşağıdaki methodları da bulunmaktadır.

  • TryGetValue()
  • TryRemove()
  • TryUpdate()
  • AddOrUpdate()
  • GetOrAdd()

Soru:

  • Aşağıdaki AddOrUpdate(): methodu ile ilgili item yok ise ekleme var ise güncelleme işlemi yapılır.
  • GetOrAdd(): Methodu ile ilgili item var ise çekilir yok ise ekleme işlemi yapılır. Aşağıdaki methodun çıktısı nasıldır? :)

Sonuç:

*ConcurrentBag<T>: Thread safe value tutan bir diğer collection tipidir. Dikkat edin bu farklı bir örnek! Aşağıda görüldüğü gibi 2 farklı thread, bir concurrentBag<int> diziyi doldurmaktadır.

  • Task t1“‘thread concurrentBag, ilgili diziye “for” ile [1 ile 6] arasındaki sayıları  doldururken, hemen altında while() ile diziye eklenen değerler ekrana yazdırılmaya başlanmıştır. Başka bir “Task t2” de, tam bu sırada aynı concurrentBag içine [6 ile 9] arasındaki sayıları yazmaya başlamıştır. Ve kodun başında while ile yazılan değer içinde, “Task t2” ile eklenen değerler de gözükmektedir. İlginç dimi. Bunu sağlayan “eventRest.WaitOne();” ile taskin sonunda 1. thread kapatılmadan bekletilip, Task t2’de  “eventRest.Set()” ile tasklerin kaldığı yerden devam etmesinin sağlanmasıdır. Concurrentbag’ler Thread safe olduğu için 2 thread’den de değerleri alıp, aynı while içinde yazabilmektedir.

Sonuç:

ConcurrentStack<T>: Gene Concurrent Collectionların Threadler ile olan ilişkisine, ConcurrentStack ile devam ediyorum. Burada 2 thread arasındaki Synchronization’ı sağlayarak, ilgili diziye eleman eklerken bir yandan da hataya düşmeden :) bir başka thread ile eleman çekeceğiz. Yani bir diziye aynı anda eleman ekleyip, aynı anda eleman çıkaracağız:)

  • “Task t1 = Task.Factory.StartNew(() =>” : 1. Task başlatılır
    • “for (int i = 0; i < maxCount; ++i)”: Belirtilen sayı kadar saydırılır.
      • “concurrentStack.Push(i)”: Sıradaki sayı diziye atılır.
      • “eventRest.Set()”: 1.Thread başlatılır.
      • “eventRest2.WaitOne()”: 2. Thread durdurulur.
  • “Task t2 = Task.Factory.StartNew(() =>” : 2. Task başlatılır.
    • “for (int i = 0; i < maxCount; ++i)” : Girilen sayı kadar saydırılır.
      • “eventRest.WaitOne()” : Bu sefer 1.Thread durdurulur.
      • “if (concurrentStack.TryPop(out int item))” : Sıradaki sayı varsa ilgili diziden çekilir. Yoksa hataya düşülmez.
      • “Console.WriteLine(item)” : Çekilen sayı ekrana yazılır.
      • “eventRest2.Set()” : 2. Task başlatılır.
  • “Task.WaitAll(t1, t2); eventRest.Close(); eventRest2.Close()” : Tüm Taskler bitinceye kadar bekletilir. “AutoResetEvent”‘ler sonlandırılır.

Sonuç: Yukarıda görüldüğü gibi iki thread “for” içinde birbirlerini durdurup, kendilerini başlatabilmektedirler. Bu da bize bir thread’in bir listeye bişeyler eklerken, bir başka thread’in aynı listeden eklenenleri çekmesine olanak vermektedir. İlk başta düşünüldüğünde hayal gibi gelen bu işlemler, Thread Safe Collectionlar ile kolaylıkla yapılabilmektedir.

Bu makalede pek fazla bilinmeyen ama çok önemli olan Thread veya Tasklar ile çalışlırken, yani Asenkron durumlarda Concurrent Collectionlardan nasıl faydalanabileceğimizden bahsettik. Bu Concurrent nesnelerin daha birçok methodu bulunmaktadır. Ama bu makalede daha çok size, asenkron programlamada işinize yarayabilecek olan methodlar anlatılmıştır.

Not: Unutulmamalıdır ki her güzel şeyin bir bedeli vardır. Bu güzel Concurrent Collectionların da kusuru performansdır. Yani kısaca işinize yaramıyacak ise, Dictionary yerine ConcurrentDictionary ya da Queue yerine ConcurrentQueue kullanılmamalıdır. Aşağıdaki resimde de görüldüğü gibi tek thread için ConcurrentDictionary performansdan dolayı kullanılmamalıdır. Çünkü normal bir Dictionary’ye göre çok daha zaman harcamaktadır.

Geldik sıkı :) bir makalenin daha sonuna. Yeni bir makalede görüşmek üzere hepinize hoşçakalın.

Kaynaklar:

Herkes Görsün:

Bunlar da hoşunuza gidebilir...

2 Cevaplar

  1. Emre Çelik dedi ki:

    Çok teşekkürler hocam çok faydalı oldu

Bir cevap yazın

E-posta hesabınız yayımlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir