Differences

This shows you the differences between two versions of the page.

--- java:java_util_concurrent [2017/03/21 11:15] – [Lock] gthanos
+++ java:java_util_concurrent [2017/03/21 14:36] (current) – [Semaphore] gthanos
@@ Line 11: / Line 11: @@
 Ένα //blocking queue// είναι μία ουρά μηνυμάτων (//queue//) στην οποία μπορούν να γράφουν ή να διαβάζουν από αυτή με ασφάλεια δύο ή περισσότερα νήματα. Όταν η ουρά είναι άδεια και ένα νήμα επιχειρήσει να διαβάσει αυτό μπλοκάρει μέχρι κάποιο άλλο νήμα να γράψει στην ουρά. Αντίστοιχα, όταν ένα νήμα επιχειρεί να γράψει σε μία γεμάτη ουρά το νήμα μπλοκάρει μέχρι να ελευθερωθεί μία θέση στην ουρά μηνυμάτων από κάποιο άλλο νήμα που διάβασε. Στο παρακάτω σχήμα βλέπετε την σχηματική αναπαράσταση ενός blocking queue.
-{{  :java:design_and_implement_a_blocking_queue-600x0.png?450  }}
+{{ :java:blockingqueue.png? }}
 Οι βασικές μέθοδοι που υποστηρίζονται από ένα //blocking queue// είναι οι εξής:
@@ Line 36: / Line 37: @@
     * **[[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#acquire()|acquire()]]:** Δέσμευει ένα permit, εφόσον αυτό είναι διαθέσιμο, διαφορετικά το νήμα μπλοκάρει. Εάν κατά την αναμονή το νήμα διακοπεί από άλλο νήμα τότε η μέθοδος "πετάει" [[https://docs.oracle.com/javase/7/docs/api/java/lang/InterruptedException.html|InterruptedException]].
     * **[[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#acquire(int)|acquire(int N)]]:** Δέσμευει Ν permits εφόσον είναι διαθέσιμα, διαφορετικά μπλοκάρει. Εάν κατά την αναμονή το νήμα διακοπεί από άλλο νήμα τότε η μέθοδος "πετάει" [[https://docs.oracle.com/javase/7/docs/api/java/lang/InterruptedException.html|InterruptedException]].
-    * **https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#acquireUninterruptibly()|acquireUninterruptibly()]]:** Δέσμευει ένα permit, εφόσον αυτό είναι διαθέσιμο, διαφορετικά το νήμα μπλοκάρει. Εάν κατά την αναμονή το νήμα διακοπεί από άλλο νήμα τότε αυτό εξακολουθεί και περιμένει.
+    * **[[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#acquireUninterruptibly()|acquireUninterruptibly()]]:** Δέσμευει ένα permit, εφόσον αυτό είναι διαθέσιμο, διαφορετικά το νήμα μπλοκάρει. Εάν κατά την αναμονή το νήμα διακοπεί από άλλο νήμα τότε αυτό εξακολουθεί και περιμένει.
     * **[[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#acquireUninterruptibly(int)|acquireUninterruptibly(int permits)]]:** Δέσμευει Ν permits εφόσον είναι διαθέσιμα, διαφορετικά μπλοκάρει. Εάν κατά την αναμονή το νήμα διακοπεί από άλλο νήμα τότε αυτό εξακολουθεί και περιμένει.
     * **[[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html#tryAcquire()|tryAcquire()]]:** Επιχειρεί να δεσμεύσει ένα permit, εφόσον αυτό είναι διαθέσιμο. Επιστρέφει **true/false** σε περίπτωση επιτυχίας/αποτυχίας.
@@ Line 60: / Line 61: @@
 Υλοποίηση του παραπάνω //interface// είναι η κλάση [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantLock.html|java.util.concurrent.locks.ReentrantLock]], η οποία επιτρέπει το πολλαπλό κλείδωμα της κλειδαριάς από το ίδιο νήμα. Το ξεκλείδωμα της κλειδαριάς πραγματοποιείται μόνον εφόσον το νήμα που κλείδωσε ξεκλειδώσει τόσες φορές όσες έχει προηγούμενα κλειδώσει. Η συγκεκριμένη κλάση δίνει την δυνατότητα του ισότιμου χρονοπρογραμματισμού των νημάτων που περιμένουν, εφόσον κληθεί [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantLock.html#ReentrantLock(boolean)|ο κατασκευαστής]] με όρισμα //true//.
 ===== ReadWriteLock =====
-Αντίστοιχα με τo interface Lock μόνο που προδιαγράφονται 2 κλειδαριές, μία για διάβασμα και μία για γράψιμο. Η διαφορά με την προηγούμενη κλειδαριά είναι ότι την κλειδαριά που επιτρέπει το διάβασμα μπορούν να την κλειδώσουν παράλληλα πολλά νήματα που θέλουν να διαβάσουν εάν δεν υπάρχει νήμα που θέλει να γράψει. Αντίστοιχα την κλειδαριά για γράψιμο μπορεί να την κλειδώσει μόνο ένα νήμα. Παράλληλα δεν επιτρέπειται η πρόσβαση και στην κλειδαριά που επιτρέπει το διάβασμα.
+Το //interface// [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReadWriteLock.html|
+ReadWriteLock]] προδιαγράφει δύο κλειδαριές, μία για διάβασμα και μία για γράψιμο. Η διαφορά με την προηγούμενη κλειδαριά είναι ότι την κλειδαριά που επιτρέπει το διάβασμα μπορούν να την κλειδώσουν παράλληλα πολλά νήματα που θέλουν να διαβάσουν εάν δεν υπάρχει νήμα που θέλει να γράψει. Αντίστοιχα την κλειδαριά που επιτρέπει το γράψιμο μπορεί να την κλειδώσει μόνο ένα νήμα και παράλληλα απαγορεύεται η πρόσβαση στην κλειδαριά που επιτρέπει το διάβασμα. Συνοπτικά:
-  * ΜΟΝΟ ένα νήμα μπορεί να γράψει.
+  * ΜΟΝΟ ένα νήμα μπορεί να γράψει. Κανένα νήμα δεν μπορεί να διαβάσει.
   * Πολλαπλά νήματα μπορούν να διαβάσουν εάν δεν υπάρχει νήμα που θέλει να γράψει.
 Το παραπάνω interface υλοποιείται μεσω της κλάσης [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.html||java.util.concurrent.locks.ReentrantReadWriteLock]]. Η κλάση αυτή περιέχει δύο άλλες εσωτερικές κλάσεις την [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.ReadLock.html|ReentrantReadWriteLock.ReadLock]] και [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.WriteLock.html|ReentrantReadWriteLock.WriteLock]]. Η πρώτη χρησιμοποιείται για διάβασμα και η δεύτερη για γράψιμο.
-===== Atomic Integer, Long, Boolean, Reference =====
-===== Παραδείγματα Χρήσης Εργαλείων Συγχρονισμού =====
-==== Blocking Queue με locks ====
+===== Atomic Integer, Long, Boolean, Reference =====
-Ας υποθέσουμε ότι έχετε το αρχείο κειμένου {{:java:largedict.txt.zip|largedict.txt}} για το οποίο θέλετε να κάνετε τα εξής. Έχετε **Ν** νήματα που διαβάζουν από το αρχείο και **Μ** νήματα που στο νέο αρχείο. Θέλετε το νέο αρχείο που θα γραφεί να είναι ίδιο με το αρχικό αρχείο που διαβάστηκε. Το αρχείο κειμένου διαβάζεται και γράφεται γραμμή-γραμμή.
+Στο πακέτο [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/package-summary.html|java.util.concurrent.atomic]] η γλώσσα παρέχει ισοδύναμους των βασικών τύπων δεδομένων οι οποίοι όμως μπορούν να προσπελαστούν από δύο ή περισσότερα νήματα χωρίς να προκύψει ασάφεια ως προς την τιμή τους. Οι τύποι αυτοί είναι οι εξής:
-Για την επικοινωνία μεταξύ αναγνωστών και εγγραφέων θα χρησιμοποιηθεί ένα blocking queue με σχετικά περιορισμένη χωρητικότητα **Κ**, όπου **K < min(M,N)**. Επίσης για την υλοποίηση του συγχρονισμού διαθέτουμε δύο κλειδαριές **α)** μία κλειδαριά η οποία βεβαιώνει ότι μόνο ένα νήμα μπορεί να γράψει στο Blocking Queue.
-Δείτε το παρακάτω παράδειγμα κώδικα που υλοποιεί τον συγχρονισμό μεταξύ των νημάτων ανάγνωσης και εγγραφής.
-<code java UtilConcurrentDemo1.java>
-import java.util.concurrent.*;
-import java.util.concurrent.locks.*;
-import java.io.*;
-public class UtilConcurrentDemo1 {
-  public static void main(String args[]) {
-    BufferedReader in;
-    FileWriter out;
-    if( args.length == 0 ) {
-      System.err.println("Insufficient number of arguments!");
-    }
-    String filename = args[0];
-    try {
-      in = new BufferedReader(new FileReader(filename));
-      out = new FileWriter(filename+".copy");
-    }
-    catch(IOException ex) {
-      ex.printStackTrace();
-      return ;
-    }
-    boolean useLocks = false;
-    ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
-    ReentrantLock readLock = new ReentrantLock();
-    ReentrantLock writeLock = new ReentrantLock();
-    FileReaderQueueWriter w1 = new FileReaderQueueWriter(in, queue, writeLock, "QueueWriter-1", useLocks);
-    FileReaderQueueWriter w2 = new FileReaderQueueWriter(in, queue, writeLock, "QueueWriter-2", useLocks);
-    FileReaderQueueWriter w3 = new FileReaderQueueWriter(in, queue, writeLock, "QueueWriter-3", useLocks);
-    FileWriterQueueReader r1 = new FileWriterQueueReader(out, queue, readLock, "QueueReader-1", useLocks);
-    FileWriterQueueReader r2 = new FileWriterQueueReader(out, queue, readLock, "QueueReader-2", useLocks);
-    FileWriterQueueReader r3 = new FileWriterQueueReader(out, queue, readLock, "QueueReader-3", useLocks);
-    w1.start();
-    w2.start();
-    w3.start();
-    r1.start();
-    r2.start();
-    r3.start();
-  }
-}
-class FileReaderQueueWriter extends Thread {
-  BufferedReader in;
-  BlockingQueue<String> queue;
-  ReentrantLock lock;
-  boolean useLocks;
-  public FileReaderQueueWriter(BufferedReader in, BlockingQueue<String> queue, ReentrantLock lock, String name, boolean useLocks) {
-    this.in = in;
-    this.queue = queue;
-    this.lock = lock;
-    setName(name);
-    this.useLocks = useLocks;
-  }
-  void printError(String msg) {
-    System.err.println("["+getName()+"] "+msg);
-  }
-  void printOut(String msg) {
-    System.out.println("["+getName()+"] "+msg);
-  }
-  public void run() {
-    int i=0; String input;
-    try {
-      while( true ) {
-        if( useLocks )
-          lock.lock();
-        input = in.readLine();
-        if( input != null) {
-          queue.put(input);
-          if( useLocks )
-            lock.unlock();
-        }
-        else {
-          if( useLocks )
-            lock.unlock();
-          break;
-        }
-      }
-    }
-    catch(IOException ex) {
-      printError("IOException while reading!");
-    }
-    catch(InterruptedException ex) {
-      printError("Queue put method interrupted.");
-    }
-    finally {
-      if( useLocks && lock.isHeldByCurrentThread() )
-        lock.unlock();
-      try {
-        in.close();
-        printError("Buffered Reader closed!");
-      } catch( IOException ex) {
-        printError("IOException while closing! Stream already closed.");
-      }
-    }
-  }
-}
-class FileWriterQueueReader extends Thread {
-  FileWriter writer;
-  BlockingQueue<String> queue;
-  ReentrantLock lock;
-  boolean useLocks;
-  public FileWriterQueueReader(FileWriter writer, BlockingQueue<String>queue, ReentrantLock lock, String name, boolean useLocks) {
-    this.writer = writer;
-    this.queue = queue;
-    this.lock = lock;
-    setName(name);
-    this.useLocks = useLocks;
-  }
-  void printError(String msg) {
-    System.err.println("["+getName()+"] "+msg);
-  }
-  void printOut(String msg) {
-    System.out.println("["+getName()+"] "+msg);
-  }
-  public void run() {
-    try {
-      String input;
-      while( true) {
-        if( useLocks )
-          lock.lock();
-        if( (input = queue.poll(500, TimeUnit.MILLISECONDS)) != null ) {
-          input += "\n";
-          writer.append( input.subSequence(0, input.length()), 0, input.length() );
-          if( useLocks )
-            lock.unlock();
-        }
-        else {
-          if( useLocks )
-            lock.unlock();
-          break;
-        }
-      }
-    } catch(IOException ex) {
-     printError("Error while writing to file!");
-    } catch(InterruptedException ex) {
-     printError("Tired of waiting. Queue is probably empty!");
-    } finally {
-      if( useLocks && lock.isHeldByCurrentThread() )
-        lock.unlock();
-      try {
-        writer.close();
-        printError("File Writer closed!");
-      } catch( IOException ex) {
-        printError("IOException while closing! Stream already closed.");
-      }
-    }
-  }
-}
-</code>
-==== Υλοποίηση μίας διασυνδεδεμένης λίστας η οποία είναι συγχρονισμένη ====
-Όπως ίσως θα γνωρίζετε η διασυνδεδεμένες λίστες που υπάρχουν στο πακέτο java.util δεν είναι συγχρονισμένες, δηλαδή δεν μπορούν να προσπελαστούν ταυτόχρονα από δύο ή περισσότερα νήματα. Προκειμένου να μπορούμε να προσπελάσουμε μία λίστα από περισσότερα του ενός νήματα δημιουργούμε την κλάση SynchronizedList η οποία χρησιμοποιεί ένα [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.ReadLock.html|java.util.concurrent.locks.ReentrantLock.ReadLock]] για διάβασμα και ένα [[http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/ReentrantReadWriteLock.ReadLock.html|ReentrantReadWriteLock.WriteLock.html]] για μεταβολή του περιεχομένου της λίστας.
-Δείτε πως διαμορφώνεται η κλάση καθώς και ένα παράδειγμα χρήσης της κλάσης αυτής.
-<code java UtilConcurrentDemo2.java>
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.locks.*;
-class SynchronizedList<E> {
-  ArrayList<E> list;
-  ReentrantReadWriteLock lock;
-  Lock rlock, wlock;
-  public SynchronizedList() {
-    list = new ArrayList<>();
-    lock = new ReentrantReadWriteLock();
-    rlock = lock.readLock();
-    wlock = lock.writeLock();
-  }
-  public SynchronizedList(int capacity) {
-    list = new ArrayList<>(capacity);
-    lock = new ReentrantReadWriteLock();
-    rlock = lock.readLock();
-    wlock = lock.writeLock();
-  }
-  public void listWriteLock() {
-    wlock.lock();
-  }
-  public void listWriteUnlock() {
-    wlock.unlock();
-  }
-  public void listReadLock() {
-    rlock.lock();
-  }
-  public void listReadUnlock() {
-    rlock.unlock();
-  }
-  public void add(E e) {
-    wlock.lock();
-    list.add(e);
-    wlock.unlock();
-  }
-  public void add(int index, E e) {
-    wlock.lock();
-    list.add(index, e);
-    wlock.unlock();
-  }
-  public boolean addAll(Collection<? extends E> c) {
-    wlock.lock();
-    boolean result = list.addAll(c);
-    wlock.unlock();
-    return result;
-  }
-  public void clear() {
-    wlock.lock();
-    list.clear();
-    wlock.unlock();
-  }
-  public boolean contains(Object o) {
-    rlock.lock();
-    boolean result = list.contains(o);
-    rlock.unlock();
-    return result;
-  }
-  public void ensureCapacity(int minCapacity) {
-    wlock.lock();
-    list.ensureCapacity(minCapacity);
-    wlock.unlock();
-  }
-  public E get(int index) {
-    rlock.lock();
-    E e = list.get(index);
-    rlock.unlock();
-    return e;
-  }
-  public int indexOf(Object o) {
-    rlock.lock();
-    int index = list.indexOf(o);
-    rlock.unlock();
-    return index;
-  }
-  public boolean isEmpty() {
-    rlock.lock();
-    boolean empty = list.isEmpty();
-    rlock.unlock();
-    return empty;
-  }
-  public int lastIndexOf(Object o) {
-    rlock.lock();
-    int lastIndex = list.lastIndexOf(o);
-    rlock.unlock();
-    return lastIndex;
-  }
-  public E remove(int index) {
-    wlock.lock();
-    E e = list.remove(index);
-    wlock.unlock();
-    return e;
-  }
-  public E set(int index, E element) {
-    wlock.lock();
-    E e = list.set(index, element);
-    wlock.unlock();
-    return e;
-  }
-  public int size() {
-    rlock.lock();
-    int lsize = list.size();
-    rlock.unlock();
-    return lsize;
-  }
-  List<E> subList(int fromIndex, int toIndex) {
-    rlock.lock();
-    List<E> newlist = list.subList(fromIndex, toIndex);
-    rlock.unlock();
-    return newlist;
-  }
-  class SyncrhonizedIterator implements Iterator<E> {
-    Iterator it;
-    Lock rlock;
-    Lock wlock;
-    public SyncrhonizedIterator(Lock readlock, Lock writelock) {
-      rlock = readlock;
-      wlock = writelock;
-      it = list.iterator();
-    }
-    public boolean hasNext() {
-      rlock.lock();
-      boolean hasnext = it.hasNext();
-      rlock.unlock();
-      return hasnext;
-    }
-    public E next() {
-      rlock.lock();
-      Object o = it.next();
-      rlock.unlock();
-      return (E)o;
-    }
-    public void remove() {
-      wlock.lock();
-      it.remove();
-      wlock.unlock();
-    }
-  }
-}
-class ListModifierThread extends Thread {
-  Random rand;
-  SynchronizedList<Integer> list;
-  public ListModifierThread(SynchronizedList<Integer> list) {
-    this.list = list;
-    rand = new Random( new Date().getTime() );
-  }
-  public void run() {
-    for(int i=0; i<1000; i++)
-      list.add( rand.nextInt(1000) );
-    list.listWriteLock();
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicBoolean.html|AtomicBoolean:]] Μία boolean μεταβλητή η οποία μπορεί να αλλάξει τιμή ατομικά.
-    while( list.size() > 0 )
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicInteger.htmlAtomicInteger:]] Μία Integer μεταβλητή η οποία μπορεί να αλλάξει τιμή ατομικά.
-      System.out.println( list.remove( list.size()-1 ) );
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicIntegerArray.html|AtomicIntegerArray:]] Ένας πίνακας από ακεραίους του οποίου τα στοιχεία μπορούν να αλλάξουν τιμή ατομικά.
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicLong.html|AtomicLong:]] Μία Long μεταβλητή η οποία μπορεί να αλλάξει τιμή ατομικά.
-    if( list.isEmpty() )
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicLongArray.html|AtomicLongArray:]] Ένας πίνακας από ακεραίους τύπου Long του οποίου τα στοιχεία μπορούν να αλλάξουν τιμή ατομικά.
-      System.out.println( this.getName() +": list is empty!");
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicReference.html|AtomicReference:]] Μία αναφορά σε αντικείμενο η οποία μπορεί να αλλάξει ατομικά.
-    else
+  * [[https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/atomic/AtomicReferenceArray.html|AtomicReferenceArray:]] Ένας πίνακας από αναφορές του οποίου τα στοιχεία μπορούν να αλλάξουν τιμή ατομικά.
-      System.out.println( this.getName() +": list is NOT empty!");
-    list.listWriteUnlock();
-  }
-}
-class ListSynchronizer {
-  public static void main(String []args) {
-    SynchronizedList<Integer> list = new SynchronizedList<>();
-    for(int i=0; i<10; i++) {
-      (new ListModifierThread(list)).start();
-    }
-  }
-}
-</code>
-<WRAP tip 80% center round>
-Αντί για τη συγχρονισμένη υλοποίηση της διασυνδεδεμένης λίστας που χρησιμοποιήσαμε μπορείτε να χρησιμοποιήσετε ένα απλό java.util.ArrayList. Αλλάξτε τον κώδικα και δείτε την συμπεριφορά (το πρόγραμμα εμφανίζει //IndexOutOfBoundException//).
-</WRAP>
+Οι βασικές μέθοδοι που διαθέτουν οι παραπάνω κλάσεις είναι οι εξής:
+  * **compareAndSet:** Η μέθοδος αλλάζει την τιμή της μεταβλητής με την προϋπόθεση ότι η υφιστάμενη τιμή ταυτίζεται με το 1ο όρισμα της μεθόδου.
+  * **get:** Επιστρέφει την τρέχουσα τιμή.
+  * **getAndSet:** Επιστρέφει την υφιστάμενη τιμή και την ανανεώνει με μία νέα.
+  * **lazySet:** Θέτει μία νέα τιμή για την μεταβλητή.
+  * **set:** Θέτει μία νέα τιμή για την μεταβλητή.