[Java] Multithread-Verarbeitung - Exklusive Steuerung

Achten Sie bei der Multithread-Verarbeitung auf die gemeinsame Nutzung von Daten

• Bei der Multithread-Verarbeitung ** Beachten Sie, ob Daten zwischen Threads geteilt werden **
• Wenn mehrere Threads gemeinsam genutzte Daten gleichzeitig verarbeiten, ** tritt eine Dateninkonsistenz auf! ** ** **

NG Beispiel

• Beispiel für das parallele Inkrementieren des im Hauptthread erstellten Wertefelds um 100.000 Threads (Ergebnisse werden voraussichtlich 100.000 sein)

//NG Beispiel
public class SynchronizedNotUse {
//Daten, die von mehreren Threads gemeinsam genutzt werden
  private int value = 0;

//Führen Sie 100.000 Threads aus
  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new SynchronizedNotUse();
    //Thread-Erstellung und Ausführung
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    //Warten Sie bis zum Ende des Threads
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value); //Jedes Mal anders
  }
  //Inkrementieren Sie das Wertefeld
  void increment() {
    this.value++;
  }
}

• Ich erwarte immer 100.000 Ergebnisse, erhalte aber jedes Mal andere Werte.
• Das Problem ist ** die Erhöhungsmethode der SynchronizedNotUse-Klasse **
• Fügen Sie in der Inkrementierungsmethode den Wert des Instanzfelds mit dem Operator ++ hinzu.
• Auf den ersten Blick scheint es, dass andere Threads nicht unterbrochen werden können, aber der ++ - Operator ist intern
• ** Den aktuellen Speicherort der Variablen abrufen, den Wert hinzufügen und das Berechnungsergebnis neu zuweisen **
• Wenn ein anderer Thread während der Verarbeitung unterbrochen wird, wird das Ergebnis nicht korrekt wiedergegeben!
• → ** Verwenden Sie den synchronisierten Befehl **!

Exklusive Steuerung mit synchronisiertem Block

• In synchronisierten Blöcken eingeschlossene Prozesse ** werden nicht mehr von mehreren Threads gleichzeitig aufgerufen **
• Auch wenn es fast zur gleichen Zeit ist, befindet sich der später angerufene im ** Wartezustand **, bis die Vorverarbeitung (Erwerb der Sperre) abgeschlossen ist.
• ** Sperren **: Besetzen Sie einen bestimmten Prozess
• ** Exklusive Kontrolle (synchrone Verarbeitung) **: Sperre, um Dateninkonsistenzen bei gleichzeitiger Ausführung zu vermeiden
• Die synchrone Verarbeitung funktioniert ordnungsgemäß ** nur, wenn die zu sperrenden Objekte dieselbe Instanz sind **
• Synchronisierung zwischen verschiedenen Sperrobjekten nicht möglich
• Wird beim Verlassen des Blocks automatisch entsperrt

public class SynchronizedUse {
  private int value = 0;

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new SynchronizedUse();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value);
  }
   //Geben Sie das zu sperrende Objekt als aktuelle Instanz an
  void increment() {
    synchronized(this) {
      this.value++;
    }
  }
}

synchronisierter Modifikator

• Wird verwendet, wenn Sie die gesamte Methode synchronisieren möchten
• Während ein Thread die synchronisierte Methode 1 ausführt, kann ein anderer Thread die synchronisierte Methode 2 nicht ausführen.
• Wird beim Beenden der Methode automatisch entsperrt
• Wenn der Compiler zwischengespeicherten Code generiert (** Optimierung **), wird synchronisierter ** aus dem ursprünglichen Speicher gelesen, sodass die Konsistenz garantiert ist **
• cf: Der Modifikator "flüchtig" kann auch nur für die Zuweisung / den Abruf von Variablen verwendet werden

synchronized void increment() {
    this.value++;
}

Explizite Sperre

• ** Explizite Sperre **: Explizite Sperrenerfassung / -freigabe erforderlich
• Implizite Sperre: Sie müssen nicht über die Erfassung / Freigabe der Sperre informiert sein (synchronisiert).
• ** ReentrantLock-Klasse **
• Wenn Sie eine explizite Sperre erhalten, ** verpacken Sie diese unmittelbar danach in einen Try-Block **
• Schließlich garantiert Block das Entsperren

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class LockBasic {
  private int value = 0;
  private final Lock lock = new ReentrantLock();

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new LockBasic();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value);
  }


  void increment() {
    //Sperrenerfassung
    lock.lock();
    try {
      this.value++;
    } finally {
      //Freischalten
      lock.unlock();
    }
  }
}

** tryLock-Methode **

• ** Sie können im Voraus prüfen, ob Sie eine Sperre erhalten können **
• Überspringen, wenn Sie nicht auf die Sperre warten müssen

  if (lock.tryLock(10,TimeUnit.SECONDS)) {
    try {
    //Verarbeitung, die ausschließlich kontrolliert werden sollte
    } finally {
      lock.unlock();
    }
  }else{
  //Was tun, wenn Sie kein Schloss bekommen können?
}

Schlossfreie exklusive Kontrolle

• ** Atomare Zuordnung, Addition / Subtraktion usw. zu einer einzelnen Variablen auf Hardwareebene **
• cf: synchronized und java.util.concurrent.locks sind exklusive Steuerelemente, die Sperren annehmen.
• ** Atomic **: Garantiert frei von Unterbrechungen
• In Java wird das Lesen und Schreiben in ** andere Variablen als Long / Double-Typ ** durch Atomic garantiert
• Da der Long / Double-Typ nicht unbedingt atomar ist, besteht die Möglichkeit einer Unterbrechung in einer Multithread-Umgebung (Datenbeschädigung).
• Sie können den ++ - Operator atomar implementieren **, indem Sie die ** getAndIncrement-Methode der AtomicInteger-Klasse verwenden!

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicBasic {
  private AtomicInteger value = new AtomicInteger();

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new AtomicBasic();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println(tb.value);
  }

  void increment() {
    value.getAndIncrement();
  }
}

Thread-Pool

• ** Von der App verwendete Threads vorbereiten und bündeln **
• Entfernen Sie gepoolte Threads nach Bedarf und kehren Sie zum Pool zurück, wenn Sie fertig sind
• ** Sparen Sie Overhead beim Spawn / Discard, indem Sie Threads wiederverwenden! ** ** **
• ** Executors-Klasse **: Thread-Pool-Generierung
• ** Methode ausführen **: Führt den angegebenen Prozess über den Thread aus
• ** Abschaltmethode **:
• Thread-Pool ** macht das Erstellen von Threads überflüssig! ** ** **

public class ThreadPool implements Runnable {
  @Override
  public void run() {
    for (var i = 0; i < 30; i++){
      System.out.println(Thread.currentThread().getName() + ":" + i);
    }
  }
}

import java.util.concurrent.Executors;

public class ThreadPoolBasic {

  public static void main(String[] args) {
    //Generieren Sie einen Thread-Pool mit 10 Threads
    var es = Executors.newFixedThreadPool(10);
    //Aufruf und Ausführung mit der Methode execute
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.shutdown();
  }
}

Planen Sie die Ausführung

• ** ScheduledThreadService-Objekt **
• Erstellt von der ** newScheduledThreadPool ** / ** newSingleScheduledExecutor ** -Methode der Executors-Klasse
• ** Planungsmethode **: Der Prozess wird nur einmal ausgeführt, nachdem die angegebene Zeit abgelaufen ist.
• ** ScheduleAtFixedRate / ScheduleWithFixedDelay-Methode **: Prozessausführung in festgelegten Zeitintervallen

import java.time.LocalDateTime;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ThreadSchedule {
  public static void main(String[] args) {
  //Vorbereitung des Fadenpools
    var sche = Executors.newScheduledThreadPool(2);
  //Planen Sie die Ausführungsregistrierung
    sche.scheduleAtFixedRate(() -> {
      System.out.println(LocalDateTime.now());
    }, 0, 5, TimeUnit.SECONDS);

    //Halten Sie den Hauptthread an, der auf die Ausführung des Zeitplans wartet
    try {
      Thread.sleep(10000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    //Fahren Sie den Thread-Pool herunter
    sche.shutdown();
  }
}

Empfangen Sie das Verarbeitungsergebnis des Threads

• ** ExecutorService ** / ** ScheduledExecutorService ** + ** Aufrufbare Schnittstelle **
• ** Callable **: Weisen Sie dem Rückgabetyp den T-Typ zu
• ** Aufrufmethode **: Code, der in der Callable-Schnittstelle ausgeführt werden soll
• ** Submit-Methode **: Führen Sie die Verarbeitung des definierten Callable-Objekts aus
• Rückgabewert ist ** Future type **: Ergebnis der asynchronen Verarbeitung
• ** get method **: Liefert den tatsächlichen Rückgabewert

//Suchen Sie eine Zufallszahl in einem anderen Thread und halten Sie einige Millisekunden an
//Zeigen Sie den Wert im Haupt-Thread an
import java.util.Random;
import java.util.concurrent.Callable;

//Callable<Integer>Weisen Sie den Integer-Typ mit zu
public class ThreadCallable implements Callable<Integer> {
  @Override
  //Code, der in der Callable-Schnittstelle ausgeführt werden soll
  public Integer call() throws Exception {
    var rnd = new Random();
    var num = rnd.nextInt(1000);
    Thread.sleep(num);
    return num;
  }
}

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;

public class ThreadCallableBasic {

  public static void main(String[] args) {
  //Thread-Ausführung
    var exe = Executors.newSingleThreadExecutor();
    var r1 = exe.submit(new ThreadCallable());
    var r2 = exe.submit(new ThreadCallable());
    var r3 = exe.submit(new ThreadCallable());

  //Thread-Ergebnisanzeige
    //Der Rückgabewert der Submit-Methode ist Future<Integer>
    try {
      System.out.println("r1: " + r1.get());
      System.out.println("r2: " + r2.get());
      System.out.println("r3: " + r3.get());
    } catch (InterruptedException | ExecutionException e) {
      e.printStackTrace();
    }
    exe.shutdown();
  }
}

Nachbearbeitung der Thread-Verarbeitung

• ** CompletableFuture-Klasse **
• Schreiben Sie die Nachbearbeitung präzise, nachdem Sie asynchrone Verarbeitungsergebnisse erhalten haben
• Führen Sie eine asynchrone Verarbeitung in Reihe durch
• Führen Sie die asynchrone Verarbeitung parallel durch
• Übergeben Sie den auszuführenden asynchronen Prozess an die ** supplyAsync-Methode **

//Suchen Sie eine Zufallszahl in einem anderen Thread und halten Sie einige Millisekunden an
//Zeigen Sie seinen Wert

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureBasic {

  public static void main(String[] args) {
    //Ausführung der asynchronen Verarbeitung
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(1000);
      heavy(num);
      return num;
    })
      //Verarbeitung nach Abschluss
      .thenAcceptAsync((result) -> {
        System.out.println(result);
      });
    System.out.println("...Nachbearbeitung...");
    heavy(7000);
  }
  //Die Dummy-Verarbeitung (schwer) unterbricht die Verarbeitung nur für die angegebene Zeit
  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Erfolg / Fehlerbehandlung sortieren

• ** whenCompleteAsync-Methode **

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureComplete {

  public static void main(String[] args) {
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(1000);
      heavy(num);
      return num;
    })
      .whenCompleteAsync((result, ex) -> {
        //Erfolg
        if (ex == null) {
          System.out.println(result);
        } else {
        //Fehler
          System.out.println(ex.getMessage());
        }
      });
    heavy(7000);
  }

  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Führen Sie mehrere asynchrone Prozesse in Reihe aus

• ** Demonstriere den wahren Wert von Complete Future **
• ** supplyAsync **: Lieferant (mit Rückgabewert ohne Argument)
• Ergebniswertgenerierung
• ** thenApplyAsync **: Funktion (mit Argument und Rückgabewert)
• Wertverarbeitung und Konvertierung
• ** thenAcceptAsync **: Consumer (mit Argumenten, kein Rückgabewert)
• Ergebnisempfangsprozess

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureSeq {

  public static void main(String[] args) {
  //Prozess 1(Zufällige Generierung)
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(5000);
      heavy(2000);
      System.out.printf("Prozess 1: %d\n", num);
      return num;
    })
  //Prozess 2(Zufällige Zeiten)
    .thenApplyAsync((data) -> {
      var result = data * 2;
      heavy(2000);
      System.out.printf("Prozess 2: %d\n", result);
      return result;
    })
  //Prozess 3(Zufälliges Doppel)
    .thenAcceptAsync((data) -> {
      var num = data * 2;
      heavy(2000);
      System.out.printf("Prozess 3: %d\n", num);
    });
    heavy(7000);
  }

  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}