java.util.concurrent

Inhalt


Thread (Task) Erzeugung

Der Erzeugung, dem Starten und Stoppen von Threads

Thread[] t = new Thread[anzahl];
for (int i = 0; i < anzahl; i++) {
    t[i] = new RunnableImpl(...,i);
    t[i].start();
}
for (int i = 0; i < anzahl; i++) {
    try { 
        t[i].join(); 
    } catch (InterruptedException e) { 
    }
}

entspricht der folgende Code

ExecutorService 
      pool = Executors.newFixedThreadPool(numCores);
Future[] f = new Future[anzahl];
for (int i = 0; i < anzahl; i++) {
    f[i] = pool.submit( new RunnableImpl(...,i) );
}
for (int i = 0; i < anzahl; i++) {
    try { 
        f[i].get(); 
    } catch (InterruptedException ignored) { 
    } catch (ExecutionException ignored) { 
    }
}
pool.shutdown();

Kritische Bereiche

Zur Behandlung von kritischen Bereichen gibt es zwei Unterpackete java.util.concurrent.locks und java.util.concurrent.atomic.

Zum Beispiel entspricht dem synchronized Konstrukt

synchronized (mutex) { ... statements ... }

die folgende Lock Konstruktion

Lock mutex  = new ReentrantLock();

mutex.lock();
try {
    ... statements ... 
} finally {
    mutex.unlock();
}

Bedingungen

Neben den schon genannten Locks gibt es eine ganze Reihe weiterer nützlicher Klassen, wie Semaphore, Barrieren und Latches.

Mit synchronized und wait():

synchronized (mutex) { 
    ... 
    while ( ? ) {
          mutex.wait();
    }
    ... 
}

synchronized (mutex) { 
    ... 
    if ( ? ) {
       mutex.notify();
    }
    ... 
}

Mit Lock und Condition:

Lock      mutex  = new ReentrantLock();
Condition   cond = mutex.newCondition();

mutex.lock();
try {
    ... 
    while ( ? ) {
          cond.await();
    }
    ...
} finally {
    mutex.unlock();
}


mutex.lock();
try {
    ... 
    if ( ? ) {
       cond.signal();
    }
    ...
} finally {
    mutex.unlock();
}

Erweiterungen im Collection Framework

Eine Reihe von Interfaces und Klassen erweitern die Collection-Tools um nützliche multithreading-Funktionalitäten, wie Queue, BlockingQueue, ConcurrentMap, ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue, Exchanger, CopyOnWriteArrayList, CopyOnWriteArraySet, ConcurrentLinkedQueue, ConcurrentHashMap.


Java 8: (Parallel-)Streams

Mit Java 8 wurden Lambda Ausdrücke und Streams eingeführt. Mit Parallel-Streams werden die multithreading Funktionalitäten wieder erweitert. Streams haben durch die Map-Reduce Funktionen von Hadoop einige Bedeutung erlangt.

Bestandteile der Streams Klassen

Beispiel: Histogram für Wörter in einem Text

Stream seq = new BufferedReader(
                         new FileReader(file)).lines(); 
Stream words = seq.flatMap( 
                     line -> Arrays.asList(line.split(" ")).parallelStream() );
int max = words.mapToInt( s -> s.length() ).max(); 
Stream<Map.Entry<String,Integer>> word1 = words.map(
      word -> new AbstractMap.SimpleEntry<String,Integer>(word,1) );

"Shuffle" Schritt

Map<String,List<Map.Entry<String,Integer>>> word2 = 
    word1.collect( Collectors.groupingBy(Map.Entry::getKey) );
Stream<Map.Entry<String,Integer>> word3 = word2.entrySet().parallelStream().map(
      e -> new AbstractMap.SimpleEntry<String,Integer>(e.getKey(), 
              e.getValue().stream().collect( Collectors.summingInt(Map.Entry::getValue)) ) );
Map<String,Integer> histo = word4.collect( 
              Collectors.toMap( e -> e.getKey(), e -> e.getValue() ) );

Ergebnis:

...
let=1, 1,=1, you=7, using=3, *=13, -=3, an=2, easy=1, standards=1, as=2, 
...

Ausblick: Android ActiveTask

Siehe ActiveTask.


© Universität Mannheim, Rechenzentrum, 2005-2017.

Heinz Kredel

Last modified: Sun Apr 23 23:25:26 CEST 2017