Java Parallel Code Sample Collection

Hier finden Sie eine Zusammenfassung der Verwendung bei der Parallelverarbeitung in Java.

Arten der Parallelisierung

Es gibt verschiedene Arten der Parallelisierung, und sie können hauptsächlich nach der Art der zu verarbeitenden Daten und der Anzahl der auszuführenden Programme klassifiziert werden.

• SPSD (Single Program Single Data): Hiermit werden Daten in einem Programm ohne Parallelisierung verarbeitet.
• SPMD (Single Program Multi Data): Verarbeitung mehrerer Daten in einem Programm
• MPSD (Multi Program Single Data): Verarbeitung von Daten mit mehreren Programmen
• MPMD (Multi Program Multi Data): Verarbeitung mehrerer Daten in mehreren Programmen

Das zu schreibende Programm hängt davon ab, welche Art von Parallelisierung Sie durchführen möchten.

SPMD-Parallelisierung

Dies wird verwendet, wenn Sie Methoden einer bestimmten Klasse gleichzeitig ausführen möchten. Im folgenden Beispiel wird der in der Aufrufmethode der TestJob-Klasse geschriebene Prozess ausgeführt. Derzeit gibt es zwei Fälle: Der eine besteht darin, die Methode zu verarbeiten und der andere darin, das Ergebnis zu erhalten.

Wenn Sie das Ergebnis nicht brauchen

Schreiben Sie als folgenden Code. Ich füge der Jobliste eine anonyme Klasse hinzu. Vergessen Sie auch nicht `` `return null```.

TestParallel.java

public class TestParallel {
	
	static final int threadNum = 4;

	public static void main(String[] args) {
		//jobs:Eine Klasse, die den Prozess beschreibt, den Sie ausführen möchten, threadpool:Thread zum Ausführen
		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 {
					//Dies muss threadsicher sein
					Thread.sleep(100L);
					return null;
				}
				
			});
		}
		try {
			threadpool.invokeAll(jobs); //Thread-Ausführung
			threadpool.shutdown();//Akzeptiere keine neuen Aufgaben
			if(threadpool.awaitTermination(5L,TimeUnit.MINUTES)){
				//Warte 5 Minuten
				System.out.println("Finish!");
			}else{
				//Endet nicht innerhalb von 5 Minuten oder endet abnormal
				System.out.println("Fail!");
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			threadpool.shutdown();
		}
	}
}

Wenn Sie Ergebnisse wollen

Schreiben Sie als folgenden Code. Dieses Mal habe ich eine Klasse für einen Job erstellt, ohne eine anonyme Klasse zu verwenden. Das Ergebnis ist vom Typ `Future <T>`, was ungefähr ein Typ ist, da eine Instanz vom Typ T als Rückgabewert zurückgegeben wird. Wenn Sie diese get () -Methode aufrufen, wartet sie, bis Sie das Ergebnis erhalten.

TestParallel.java

public class TestParallel {
	
	static final int threadNum = 4;

	public static void main(String[] args) {
		//jobs:Eine Klasse, die den Prozess beschreibt, den Sie ausführen möchten, threadpool:Thread zum Ausführen
		Collection<Callable<String>> jobs = new ArrayList<Callable<String>>();
		ExecutorService threadpool = Executors.newFixedThreadPool(threadNum);
		List<Future<String>> futures; //Speichern Sie das Ergebnis
		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); //Thread-Ausführung
			threadpool.shutdown();//Akzeptiere keine neuen Aufgaben
			for(int i = 0; i < threadNum; i++){
				//Ergebnisse bekommen
				result.add(futures.get(i).get());
			}
			for(String s: result){
				//Ergebnisse anzeigen
				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 {
		//Dies muss threadsicher sein
		Thread.sleep(100L);
		return "thread " + Integer.toString(i);
	}
}

Verwenden Sie Java 8 Stream

Zusätzlich zu dem bisher eingeführten Executor Service ist die Parallelverarbeitung mit Stream ab Java 8 möglich geworden.

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 -> {
		   //Diese Zwischenoperation muss threadsicher sein
			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();
		}
	}
}

Wenn Sie das Ergebnis erhalten möchten, weisen Sie es der in main oder so deklarierten Liste zu. In diesem Fall muss jedoch die Operation zum Zeitpunkt der Zuweisung ausschließlich gesteuert werden, und welcher Thread ist das Ergebnis? Ich weiß es nicht. Wenn Sie wissen, wie man es wie Future bekommt, würde ich es begrüßen, wenn Sie mir beibringen könnten ...

Parallelisierung von MPSD und MPMD

Lassen Sie uns gehorsam einen Thread erstellen. Ich denke, MPSD ist ziemlich schwierig, deshalb werde ich hier den Code am Beispiel von MPMD schreiben. Es ist ein Prozess, "a" am Ende für die Abfrage 1-Gruppe und "b" am Ende für die Abfrage 2-Gruppe hinzuzufügen.

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

Wenn Sie den Thread beenden möchten, können Sie ihn beenden, indem Sie eine Ausnahme mit `Thread # interrupt ()` auslösen und ihn im Thread abfangen. `Thread # stop ()` ist veraltet und sollte nicht viel verwendet werden. Ich denke, dass die Methode zum Erstellen von Thread implementiert wurde, wenn Sie GUI usw. (wahrscheinlich) erstellt haben. Verwenden Sie außerdem "synchronisiert", um die ausschließliche Kontrolle über den Teil zu ermöglichen, der Konflikte verursacht.

Parallel verarbeiten

Bisher habe ich überlegt, parallel mit mehreren Threads zu verarbeiten. Sie können jedoch auch einen externen Prozess erstellen und parallel ausführen. Der folgende Code entpackt hoge.tar.gz. Hier wartet die waitiFor-Methode auf das Ende, es ist jedoch auch möglich, unverändert fortzufahren.

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

Zusammenfassung

Ich persönlich mochte die Java 8-Notation sehr prägnant. Grundsätzlich ist die Parallelisierung einfacher, wenn Sie Daten parallel parallelisieren, wenn (natürlich) kein Konflikt besteht. Ich würde es auch begrüßen, wenn Sie mir sagen könnten, ob es andere Methoden wie diese gibt.