Faites attention au partage de données dans le traitement multi-thread

Dans le traitement multi-thread ** Sachez si les données sont partagées entre les threads **
Si plusieurs threads traitent des données partagées en même temps, ** une incohérence des données se produira! ** **

Exemple NG

Exemple d'incrémentation du champ de valeur préparé dans le thread principal en parallèle de 100 000 threads (les résultats devraient être de 100 000)

//Exemple NG
public class SynchronizedNotUse {
//Données partagées par plusieurs threads
  private int value = 0;

//Exécuter 100 000 threads
  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new SynchronizedNotUse();
    //Création et exécution de threads
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    //Attendez la fin du fil
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value); //Différent à chaque fois
  }
  //Incrémenter le champ de valeur
  void increment() {
    this.value++;
  }
}

J'attends toujours 100 000 résultats, mais j'obtiens des valeurs différentes à chaque fois.
Le problème est ** la méthode d'augmentation de la classe SynchronizedNotUse **
Dans la méthode d'incrémentation, ajoutez la valeur du champ d'instance avec l'opérateur ++.
À première vue, il semble que les autres threads ne puissent pas interrompre, mais l'opérateur ++ est en interne
** Obtenez l'emplacement actuel de la variable, ajoutez la valeur, réaffectez le résultat du calcul **
Si un autre thread s'interrompt pendant le traitement, le résultat ne sera pas reflété correctement!
→ ** Utilisez la commande synchronisée **!

Contrôle exclusif avec bloc synchronisé

Les processus inclus dans des blocs synchronisés ** ne seront plus appelés par plusieurs threads en même temps **
Même si c'est presque la même heure, celui appelé plus tard sera en ** état d'attente ** jusqu'à la fin du prétraitement (acquisition du verrou)
** Verrouiller **: occuper un processus spécifique
** Contrôle exclusif (traitement synchrone) **: Verrouillage pour éviter les incohérences des données dues à l'exécution simultanée
Le traitement synchrone fonctionne correctement ** uniquement lorsque les objets à verrouiller sont la même instance **
Impossible de synchroniser entre différents objets de verrouillage
Déverrouille automatiquement à la sortie du bloc

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);
  }
   //Spécifiez l'objet à verrouiller comme instance actuelle
  void increment() {
    synchronized(this) {
      this.value++;
    }
  }
}

modificateur synchronisé

Utilisé lorsque vous souhaitez synchroniser l'ensemble de la méthode
Lorsqu'un thread exécute la méthode synchronisée 1, un autre thread ne peut pas exécuter la méthode synchronisée 2.
Déverrouille automatiquement en quittant la méthode
Lorsque le compilateur génère du code mis en cache (** optimisation **), synchronized ** est lu à partir de la mémoire d'origine, donc la cohérence est garantie **
cf: Le modificateur volatile peut également être utilisé uniquement pour l'affectation / la récupération de variables

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

Verrou explicite

** Verrou explicite **: acquisition / libération explicite du verrou requis
Verrou implicite: vous n'avez pas à être conscient de l'acquisition / libération du verrou (synchronisé)
** Classe ReentrantLock **
Si vous obtenez un verrou explicite, ** immédiatement après cela, enveloppez-le dans un bloc try **
Enfin le bloc garantit le déverrouillage

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() {
    //Verrouiller l'acquisition
    lock.lock();
    try {
      this.value++;
    } finally {
      //ouvrir
      lock.unlock();
    }
  }
}

Méthode tryLock

** Vous pouvez vérifier à l'avance si vous pouvez obtenir une serrure **
Sautez si vous n'avez pas besoin d'attendre le verrouillage

  if (lock.tryLock(10,TimeUnit.SECONDS)) {
    try {
    //Traitement qui devrait être contrôlé exclusivement
    } finally {
      lock.unlock();
    }
  }else{
  //Que faire si vous ne pouvez pas obtenir de cadenas
}

Contrôle exclusif sans verrouillage

** Affectation atomique, addition / soustraction, etc. à une seule variable au niveau matériel **
cf: synchronized et java.util.concurrent.locks sont des contrôles exclusifs qui supposent des verrous.
** Atomic **: garanti sans interruption
En Java, la lecture et l'écriture dans ** des variables autres que le type long / double ** sont garanties par atomic
Le type long / double n'étant pas nécessairement atomique, il existe une possibilité d'interruption dans un environnement multi-thread (corruption de données)
Vous pouvez implémenter l'opérateur ++ de manière atomique ** en utilisant la méthode ** getAndIncrement de la classe AtomicInteger!

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();
  }
}

Pool de threads

** Préparez et regroupez les threads utilisés par l'application à l'avance **
Supprimez les threads regroupés au besoin et revenez au pool une fois terminé
** Économisez des frais généraux sur le spawn / discard en réutilisant les threads! ** **
** Classe Executors **: génération de pool de threads
** execute method **: Exécute le processus spécifié via le thread
** méthode d'arrêt **:
Thread pool ** élimine le besoin d'être conscient de la création de thread! ** **

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) {
    //Générer un pool de threads avec 10 threads
    var es = Executors.newFixedThreadPool(10);
    //Appel et exécution avec la méthode d'exécution
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.shutdown();
  }
}

Planifier l'exécution

** Objet ScheduledThreadService **
Généré par la méthode ** newScheduledThreadPool ** / ** newSingleScheduledExecutor ** de la classe Executors
** méthode de planification **: le processus est exécuté une seule fois après l'écoulement du temps spécifié.
** Méthode scheduleAtFixedRate / scheduleWithFixedDelay **: Exécution du processus à des intervalles de temps spécifiés

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

public class ThreadSchedule {
  public static void main(String[] args) {
  //Préparation du pool de threads
    var sche = Executors.newScheduledThreadPool(2);
  //Enregistrement de l'exécution du calendrier
    sche.scheduleAtFixedRate(() -> {
      System.out.println(LocalDateTime.now());
    }, 0, 5, TimeUnit.SECONDS);

    //Suspendre le thread principal en attendant que le programme s'exécute
    try {
      Thread.sleep(10000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    //Arrêtez le pool de threads
    sche.shutdown();
  }
}

Recevoir le résultat du traitement du thread

** ExecutorService ** / ** ScheduledExecutorService ** + ** Interface appelable **
** Callable **: attribuer le type T au type de retour
** méthode d'appel **: Code à exécuter dans l'interface Callable
** submit method **: exécute le traitement de l'objet Callable défini
La valeur de retour est ** Type futur **: résultat du traitement asynchrone
** get method **: Obtenez la valeur de retour réelle

//Trouvez un nombre aléatoire dans un autre thread et faites une pause pendant quelques millisecondes
//Afficher la valeur dans le thread principal
import java.util.Random;
import java.util.concurrent.Callable;

//Callable<Integer>Attribuer un type entier avec
public class ThreadCallable implements Callable<Integer> {
  @Override
  //Code à exécuter dans l'interface appelable
  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) {
  //Exécution de thread
    var exe = Executors.newSingleThreadExecutor();
    var r1 = exe.submit(new ThreadCallable());
    var r2 = exe.submit(new ThreadCallable());
    var r3 = exe.submit(new ThreadCallable());

  //Affichage des résultats du fil
    //La valeur de retour de la méthode d'envoi est 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();
  }
}

Post-traitement du traitement des threads

** Classe CompletableFuture **
Écrire de manière concise le post-traitement après avoir reçu les résultats du traitement asynchrone
Effectuer un traitement asynchrone en série
Effectuer un traitement asynchrone en parallèle
Passez le processus asynchrone à exécuter à la ** méthode supplyAsync **

//Trouvez un nombre aléatoire dans un autre thread et faites une pause pendant quelques millisecondes
//Montrez sa valeur

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

public class FutureBasic {

  public static void main(String[] args) {
    //Exécution du traitement asynchrone
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(1000);
      heavy(num);
      return num;
    })
      //Traitement après achèvement
      .thenAcceptAsync((result) -> {
        System.out.println(result);
      });
    System.out.println("...Tout post-traitement...");
    heavy(7000);
  }
  //Le traitement factice (lourd) suspend le traitement uniquement pendant la durée spécifiée
  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Tri des succès / gestion des erreurs

** Méthode whenCompleteAsync **

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) -> {
        //Succès
        if (ex == null) {
          System.out.println(result);
        } else {
        //Échec
          System.out.println(ex.getMessage());
        }
      });
    heavy(7000);
  }

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

Exécuter plusieurs processus asynchrones en série

** Démontrez la vraie valeur de Complete Future **
** supplyAsync **: Supplier (avec valeur de retour sans argument)
Génération de valeur de résultat
** thenApplyAsync **: Fonction (avec argument et valeur de retour)
Traitement de la valeur et conversion
** thenAcceptAsync **: Consumer (avec arguments, pas de valeur de retour)
Processus de réception des résultats

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

public class FutureSeq {

  public static void main(String[] args) {
  //Processus 1(Génération aléatoire)
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(5000);
      heavy(2000);
      System.out.printf("Processus 1: %d\n", num);
      return num;
    })
  //Processus 2(Temps aléatoires)
    .thenApplyAsync((data) -> {
      var result = data * 2;
      heavy(2000);
      System.out.printf("Processus 2: %d\n", result);
      return result;
    })
  //Processus 3(Double aléatoire)
    .thenAcceptAsync((data) -> {
      var num = data * 2;
      heavy(2000);
      System.out.printf("Processus 3: %d\n", num);
    });
    heavy(7000);
  }

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

[Java] Traitement multi-thread - Contrôle exclusif

Faites attention au partage de données dans le traitement multi-thread

Exemple NG

Contrôle exclusif avec bloc synchronisé

modificateur synchronisé

Verrou explicite

** Méthode tryLock **

Contrôle exclusif sans verrouillage

Pool de threads

Planifier l'exécution

Recevoir le résultat du traitement du thread

Post-traitement du traitement des threads

Tri des succès / gestion des erreurs

Exécuter plusieurs processus asynchrones en série

Méthode tryLock