Thread Safety HashMap

Merhaba,
Bu yazida thread safe hashmap ile ilgili ornek yapacagiz.
Once thread safe olmayan bir ornek yapalim.

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class UnsafeMap {
 public static void main(String[] args) throws InterruptedException {
  Map cricketMap = new HashMap();
  cricketMap.put("Australia", 349);
  cricketMap.put("India", 257);
  ExecutorService executorService = Executors.newFixedThreadPool(10);
  Runnable task = new Runnable() {
   @Override
   public void run() {
    incrementScore(cricketMap, "India");
   }
  };
  for(int i=0; i<100 :="" ap="" cricketmap.get="" executorservice.awaittermination="" executorservice.shutdown="" executorservice.submit="" i="" inal="" incrementscore="" india="" integer="" is="" ndia="" of="" private="" score="" static="" system.out.println="" task="" timeunit.seconds="" tring="" void=""> teamMap, String team) {
  Integer score = teamMap.get(team);
  teamMap.put(team, score +1);
 }
}

Yukaridaki ornekte India nin score unu 100 arttirdik. Sonucun 258 olmasini bekliyoruz ama her seferinde baska sonuc cikiyor.

Final score of India is : 353

Final score of India is : 352

gibi. Neden ?

Cunku 

since multiple threads try to modify the HashMap concurrently, the change done by one thread gets overridden by some other thread, and the output becomes non-deterministic.

Bunu onlemek icinse ornegimizi thread safe yapalim. Yani sadece ayni anda tek bir thread erisebilsin. Bunun icin yapacagimiz iki sey sunlar olmali :

• Use the Collections.synchronizedMap() method to obtain a synchronized view of the HashMap.
• Write the increment logic inside a synchronized block.

Son halde kodumuz su sekilde olur:

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Safemap {
 public static void main(String[] args) throws InterruptedException {
  Map cricketScore = Collections.synchronizedMap(new HashMap());  
  cricketScore.put("Turkey", 52);
  cricketScore.put("Germany", 170);   
  //we are creating thread pool with size 10
  ExecutorService executorService = Executors.newFixedThreadPool(10);
  
  Runnable task = new Runnable() {   
   @Override
   public void run() {
    incrementScore(cricketScore, "Turkey");    
   }
  };
  for(int i=0; i<100 :="" ap="" cricketscore.get="" executorservice.awaittermination="" executorservice.shutdown="" executorservice.submit="" i="" inal="" incrementscore="" integer="" is="" of="" private="" score="" static="" system.out.println="" task="" timeunit.seconds="" tring="" turkey="" urkey="" void=""> scoreMap, String team) {
  synchronized (scoreMap) {
   Integer score = scoreMap.get(team);
   scoreMap.put(team, score +1);
  }
 }
}

Turkiyenin score unu 100 arttirdik. Threas safe oldugu icin her zaman 152 donecektir.

Final score of Turkey is : 152

Interthread Communication

Asagidaki ornegi incelemeye baslayalim.

Oncelik A adli bir classimiz var. Setleme ve getleme yapmasi icin put ve get methodlarimiz var. Her 2 methodumuz da degerlerimizi anlik olarak yazdirdik.

Daha sonra ise Producer ve Consumer adli classlarimiz var. Producer deger setlemek Consumer ise getlemek icin. Producer icinde sonsuz bir dongu var ve her seferinde degeri +1 i olarak setliyor.

Consumer icinde de sonsuz bir dongu var ve degeri get ediyor. Bu islemleri run() methodu icinde yapiyoruz. Bu run methodunun cagirilmasi icinse Thread e ihtiyacimiz var. O yuzden constuctor icinde thread olusturuyoruz ve start methodunu cagiriyoruz cunku start() methodu run i cagiracaktir.

class A{
 int num;
 // we need two methods here. one for set value and one for get the value
 public void put(int num) {
  System.out.println("Put : "+ num);
  this.num = num;
 } 
 public void get() {
  System.out.println("Get : "+ num);
 }
}
class Producer implements Runnable{ 
 A a;  
 public Producer(A a){
  this.a = a;
  // for Producer to call run method we need to create Thread. with start method, it will call run  
  Thread t = new Thread(this, "Producer");
  t.start();
 }
 public void run() {
  int i=0;
  // we have infinite loop here. every loop we set the value as i++
  while(true) {
   a.put(i++);
   try { Thread.sleep(1000); } catch(Exception e) {}
  }
 }
}
class Consumer implements Runnable{
 A a; 
 public Consumer(A a) {
  this.a = a;
  Thread t = new Thread(this, "Consumer");
  t.start();
 } 
 public void run() {
  while(true) {
   a.get();
   try { Thread.sleep(1000); } catch(Exception e) {}
  }
 }
}
public class ThreadCommunication {
 public static void main(String[] args) {
  A a = new A();
  new Producer(a);
  new Consumer(a);
 }}

Bunu calistirdigimiz da ise aldigimiz sonuca bakalim:
Put : 0
Get : 0
Get : 0
Put : 1
Put : 2
Get : 1
Put : 3
Get : 2
Put : 4
Get : 3
Put : 5
Get : 4

Gordugumuz gibi 0 icin 2 kez Get calismis. Bizim istedigimiz bu degildi. once setlesin sonra getlesin ve bu sirayla olsundu.

Bunun icin boolean bir degisken olusturacagiz ve default olarak false a setlicez.

put() icinde kontrol edelim. eger true ise yani zaten setlenmisse bizim setlememize gerek yok demektir. Bu durumda Consumer in degeri consume etmesini beklicez. Bunu wait() methodu ile yapicaz. Simdi su bilgiyi hatirlayalim.

Sleep() methodu kullandigimizda thread wait state ine gecer ve o sure bitince tekrar runnable olur. ama wait() ile oyle bir durum yok. Bekler surekli. Bu durumda bunu birilerine haber verelim ki bekledigini anlasin. Bunu notify() methodu ile yapariz.

Bunun icin haber vermemiz gereken consumer dur. Bunun icin setlendikten sonra notify() cagiriyoruz ve notify methodu consumer i notify edicek.

Simdi ayni islemleri get icin dusunelim. Eger valueSet degeri false gelirse deger setlenmemis demektir. Bu durumda setlenmemis degeri okuyamayiz. Beklememiz lazim. Kimi ? Producer i. yine wait cagirdik. Eger deger setlenmis ile okuyalim.

Ayrica wait() cagirdigimiz tum methodlarin synchronized olmasi zorundadir.

Kodumuzun son hali asagidaki gibi olacaktir.

Ayrica sleep surelerini de degistirdim. 0.5 saniyede setliyor ve consumer 2 saniyede okuyor. bu durumda consumer in okumasini beklicez, o okumadan asla setlemeye gecmicek.

class A{
 int num;
 // to keep track the value we created a boolean value
 boolean valueSet = false;
 // we need two methods here. one for set value and one for get the value
 public synchronized void put(int num) {
  while(valueSet) {
   // eger valueSet true ise deger zaten setlenmistir.
   // bir daha setlemeye gerek yok.
   // o yuzden burada beklicez yani wait cagiricaz
   // with wait method it will wait for consumer to consume the value
   
   try { wait(); } catch(Exception e) {}
  }  
  System.out.println("Put : "+ num);
  this.num = num;
  valueSet = true;
  notify();
 } 
 public synchronized void get() {
  while(!valueSet) {
   try { wait(); } catch(Exception e) {}
  }
  System.out.println("Get : "+ num);
  valueSet = false;
  notify();
 }
}
class Producer implements Runnable{ 
 A a;  
 public Producer(A a){
  this.a = a;
  // for Producer to call run method we need to create Thread. with start method, it will call run
  Thread t = new Thread(this, "Producer");
  t.start();
 }
 public void run() {
  int i=0;
  // we have infinite loop here. every loop we set the value as i++
  while(true) {
   a.put(i++);
   try { Thread.sleep(500); } catch(Exception e) {}
  }
 }
}
class Consumer implements Runnable{
 A a; 
 public Consumer(A a) {
  this.a = a;
  Thread t = new Thread(this, "Consumer");
  t.start();
 } 
 public void run() {
  while(true) {
   a.get();
   try { Thread.sleep(2000); } catch(Exception e) {}
  }
 }
}
public class ThreadCommunication {
 public static void main(String[] args) {
  A a = new A();
  new Producer(a);
  new Consumer(a);
 }}

Output:
Put : 0
Get : 0
Put : 1
Get : 1
Put : 2
Get : 2
Put : 3
Get : 3
Put : 4
Get : 4
Put : 5
Get : 5
Put : 6