Collection d'exemples de code parallèle Java

Voici un résumé de ce qu'il faut utiliser lors de l'exécution d'un traitement parallèle en Java.

Types de parallélisation

Il existe différents types de parallélisation en un mot, et ils peuvent être principalement classés en fonction du type de données à traiter et du nombre de programmes à exécuter.

• SPSD (Single Program Single Data): Il s'agit de traiter une donnée dans un programme sans aucune parallélisation.
• SPMD (Single Program Multi Data): Traitement de plusieurs données dans un seul programme
• MPSD (Multi Program Single Data): traitement d'une donnée avec plusieurs programmes
• MPMD (Multi Program Multi Data): Traitement de plusieurs données avec plusieurs programmes

Le programme à écrire dépend du type de parallélisation que vous souhaitez effectuer.

Parallélisation SPMD

Ceci est utilisé lorsque vous souhaitez exécuter des méthodes d'une certaine classe en même temps. Dans l'exemple suivant, le processus écrit dans la méthode d'appel de la classe TestJob est exécuté. À ce stade, il y a deux cas, l'un consiste à traiter la méthode et l'autre à obtenir le résultat.

Si vous n'avez pas besoin du résultat

Écrivez comme le code suivant. J'ajoute à la liste d'emplois en utilisant une classe anonyme. N'oubliez pas non plus `` return null ''.

TestParallel.java

public class TestParallel {
	
	static final int threadNum = 4;

	public static void main(String[] args) {
		//jobs:Une classe qui décrit le processus que vous souhaitez exécuter, threadpool:Fil à exécuter
		Collection<Callable<Void>> jobs = new ArrayList<Callable<Void>>();
		ExecutorService threadpool = Executors.newFixedThreadPool(threadNum);
		for(int i = 0; i < threadNum; i++){
			jobs.add(new Callable<Void>(){
				@Override
				public Void call() throws Exception {
					//Cela doit être thread-safe
					Thread.sleep(100L);
					return null;
				}
				
			});
		}
		try {
			threadpool.invokeAll(jobs); //Exécution de thread
			threadpool.shutdown();//N'acceptez pas de nouvelles tâches
			if(threadpool.awaitTermination(5L,TimeUnit.MINUTES)){
				//Attends 5 minutes
				System.out.println("Finish!");
			}else{
				//Ne se termine pas dans les 5 minutes ou se termine anormalement
				System.out.println("Fail!");
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			threadpool.shutdown();
		}
	}
}

Si vous voulez des résultats

Écrivez comme le code suivant. Cette fois, j'ai créé une classe pour un travail sans utiliser de classe anonyme. Le résultat est de type `` Future '', qui est à peu près le type qui renvoie une instance de type T comme valeur de retour. Si vous appelez cette méthode get (), elle attendra que vous obteniez le résultat.

TestParallel.java

public class TestParallel {
	
	static final int threadNum = 4;

	public static void main(String[] args) {
		//jobs:Une classe qui décrit le processus que vous souhaitez exécuter, threadpool:Fil à exécuter
		Collection<Callable<String>> jobs = new ArrayList<Callable<String>>();
		ExecutorService threadpool = Executors.newFixedThreadPool(threadNum);
		List<Future<String>> futures; //Stocker le résultat
		List<String> result = new ArrayList<String>();
		for(int i = 0; i < threadNum;i++){
			TestJob t = new TestJob(i);
			jobs.add(t);
		}
		try {
			futures = threadpool.invokeAll(jobs); //Exécution de thread
			threadpool.shutdown();//N'acceptez pas de nouvelles tâches
			for(int i = 0; i < threadNum; i++){
				//Obtenir des résultats
				result.add(futures.get(i).get());
			}
			for(String s: result){
				//Voir les résultats
				System.out.println(s);
			}
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
			threadpool.shutdown();
		}
	}
}

class TestJob implements Callable<String>{
	private int i;
	
	public TestJob(int i){
		this.i = i;
	}
	
	@Override
	public String call() throws Exception {
		//Cela doit être thread-safe
		Thread.sleep(100L);
		return "thread " + Integer.toString(i);
	}
}

Utiliser Java 8 Stream

En plus du service Executor introduit jusqu'à présent, le traitement parallèle à l'aide de Stream est devenu possible à partir de Java 8.

TestParallel.java

public class TestParallel {
	
	public static void main(String[] args) {
		List<TestJob> jobs = new ArrayList<TestJob>(); 
		for(int i = 0; i < 4; i++){
			jobs.add(new TestJob(i));
		}
		jobs.parallelStream().forEach(x -> {
		   //Cette opération intermédiaire doit être thread-safe
			x.doJob();
		});
	}
}

class TestJob{
	private int i;
	
	public TestJob(int i){
		this.i = i;
	}
	
	public void doJob() {
		try {
			Thread.sleep(100L);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

Si vous voulez obtenir le résultat, l'assignez-vous à la liste déclarée dans main ou quelque chose ... Cependant, dans ce cas, l'opération au moment de l'affectation doit être exclusivement contrôlée, et quel thread est le résultat? Je ne sais pas. Si vous savez comment l'obtenir comme Future, je vous serais reconnaissant si vous pouviez m'apprendre ...

Parallélisation de MPSD et MPMD

Faisons un fil avec obéissance. Je pense que MPSD est assez difficile, alors j'écrirai ici le code en utilisant MPMD comme exemple. C'est un processus pour ajouter "a" à la fin pour le groupe de requête 1 et "b" à la fin pour le groupe de requête 2.

TestParallel.java

public class TestParallel {
	
	public static void main(String[] args) {
		List<String> query1List = new ArrayList<String>();
		List<String> res1 = new ArrayList<String>();
		List<String> query2List = new ArrayList<String>();
		List<String> res2 = new ArrayList<String>();
		for(int i = 0; i < 10; i++){
			query1List.add(Integer.toString(i));
			query2List.add(Integer.toString(i+10));
		}
		Thread t1 = new Thread(new Runnable(){
			@Override
			public void run() {
				query1List.stream().forEach(x -> {
					 res1.add(x + "a");
				});
			}
		});
		Thread t2 = new Thread(new Runnable(){
			@Override
			public void run() {
				query2List.stream().forEach(x ->{
					res2.add(x + "b");
				});
			}
		});
		t1.start();
		t2.start();
		try {
			t1.join();
			t2.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

Si vous souhaitez terminer le thread, vous pouvez le terminer en levant une exception avec Thread # interruption () '' et en l'attrapant dans le thread. Thread # stop () '' est obsolète et ne devrait pas être beaucoup utilisé. Je pense que la méthode de création de Thread a été implémentée si vous avez créé une interface graphique, etc. (probablement). De plus, utilisez `` synchronized '' comme approprié pour le contrôle exclusif de la partie qui provoque le conflit.

Processus parallèle

Jusqu'à présent, nous avions pensé au traitement parallèle avec plusieurs threads, mais vous pouvez également créer un processus externe et l'exécuter en parallèle. Le code ci-dessous décompresse hoge.tar.gz. Ici, la méthode waitiFor attend la fin, mais il est également possible de procéder telle quelle sans attendre.

TestParallel.java

public class TestParallel {
		ProcessBuilder pb = new ProcessBuilder(new String[] {"tar", "zxvf", "hoge.tar.gz"});
		try {
			Process p = pb.start();
			p.waitFor();
		} catch (IOException | InterruptedException e) {
			e.printStackTrace();
		}
}

Résumé

Personnellement, j'ai aimé la notation Java 8 car elle était très concise. Fondamentalement, la parallélisation est plus facile si vous parallélisez des données en parallèle là où il n'y a pas de conflit (naturellement). J'apprécierais également que vous me disiez s'il existe d'autres méthodes comme celle-ci.