Je viens d'écrire et de comprendre le serveur gRPC, mais je ne sais pas grand-chose à ce sujet, donc je ne peux pas avancer. Cependant, essayez-le lentement à votre manière. Le deuxième sujet concerne les interfaces «Future» et «Callable». Sans comprendre cela, «gRPC» ne peut pas être compris. Cette zone a un parfum qui semble être similaire au mécanisme de C # Task et async / await, donc en tant que thème
Je souhaite implémenter ** FanOut / FanIn ** en Java.
L'interface est telle qu'elle renvoie simplement une valeur de retour de type T.
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}
Créons une classe d'implémentation qui implémente ceci. C'est la partie qui fonctionne en parallèle. Puisqu'il est supposé que plusieurs threads s'exécuteront en parallèle, l'id de CurrentThread est affiché. Une chose qui m'intéresse ici est, est-ce que c'était correct d'utiliser Thread.sleep ()? Je dis ça. Avec async / await en C #, c'est illégal, car les appels async n'occupent pas toujours Thread, il n'est donc pas pratique de mettre Thread en veille. Même si je lis Java Delay - 4 Ways to Add Delay in Java, il n'y a pas de commentaires particuliers, alors utilisons-le pour l'instant.
Activity
public class Activity implements Callable<String> {
public String call() throws Exception {
Thread currentThread = Thread.currentThread();
for (int i = 0; i < 100; i++) {
Thread.sleep(100);
System.out.println("[" + currentThread.getId() + "] Activity: " + i);
}
System.out.println("[" + currentThread.getId() +"] Done");
return "Thread [" + currentThread.getId() + "] has done.";
}
}
Maintenant, créez une classe qui exécute ceci. Je veux traiter les activités en parallèle, et lorsque tout est terminé, je veux afficher le résultat et terminer.
Pour Java, il semble utiliser quelque chose appelé ʻExecutorService. Plusieurs méthodes ont été définies dans ʻExecutors, et celles que nous utilisons maintenant sont définies pour créer un nombre fixe de pools de threads, créer un seul pool de threads, un cache, etc. Dans ce cas, il dispose de 5 pools de threads. Donc, contrairement à C #, il occupe le thread de temps en temps, est-il donc normal d'utiliser Thread.sleep ()?
Future interface
Future est une image de l'équivalent de «Task» en C #. Cet objet est renvoyé en tant que valeur de retour lorsque le thread est exécuté et que le travail parallèle est démarré. L'émission de la méthode get se bloque jusqu'à ce que le thread termine le traitement.
public interface Future<V>
{
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
La méthode ʻexecutor.submitcrée ceFuture. Si vous passez ʻActivity plus tôt, le traitement sera exécuté en parallèle. Stockez l'objet de Future dans la liste. Ensuite, attendez à ʻexecutor.awaitTermination. Le résultat final est obtenu et affiché par la méthode get ()`. Cependant, il y a un problème ici.
public class FanOutFanIn {
public void execute() throws InterruptedException, ExecutionException {
ExecutorService executor = Executors.newFixedThreadPool(5);
List<Future<String>> results = new ArrayList<>();
for (int i = 0; i < 10 ; i++) {
Future<String> result = executor.submit(new Activity());
results.add(result);
}
executor.awaitTermination(30, TimeUnit.SECONDS);
for (int i = 0; i < results.size(); i++) {
System.out.println(results.get(i).get());
}
}
}
Mon image d'origine était que tout le traitement était terminé ou que le résultat de l'exécution était affiché lorsque le délai d'expiration s'est produit, mais ce n'est pas le cas. ʻAwaitTermination` continue de bloquer même après que tout le traitement est terminé, et avec ce code, il ne passe au bloc suivant que 30 secondes.
CompletableFuture
Par conséquent, il semble y avoir quelque chose qui s'appelle «Completabled Future». L'implémentation réelle de l'interface est longue, je vais donc l'omettre, mais c'est une classe qui implémente les interfaces Future et CompletionStage. Du «futur», il a un «achèvement» définitif. Cela semble être utile à diverses fins, mais cela semble également convenir aux scénarios FanOut / FanIn. Écrivez celui ci-dessus qui fonctionne comme vous le souhaitez.
«CompletableFuture» est implémenté par «Runnable» car «Callable» ne peut pas être utilisé. C'est un peu maladroit et je pense que cela pourrait être mieux écrit, mais Runnable ne peut pas renvoyer une valeur de retour. Si je voulais me comporter de la même manière que le code précédent (renvoyer une chaîne), c'est une douleur, mais j'ai implémenté une méthode appelée getResult. Ouaip. Il doit y avoir une meilleure façon de l'écrire. Veuillez laisser un commentaire.
public class RunnableActivity implements Runnable {
private String result;
@Override
public void run() {
Thread currentThread = Thread.currentThread();
for (int i = 0; i < 100; i++) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("[" + currentThread.getId() + "] Activity: " + i);
}
System.out.println("[" + currentThread.getId() +"] Done");
this.result = "Thread [" + currentThread.getId() + "] has done.";
}
public String getResult() {
return this.result;
}
}
J'écris runAsync par moi-même, mais je l'écris moi-même pour forcer la valeur de retour de Runnable à revenir. Ouaip. Il doit y avoir un meilleur moyen w. C'est similaire au précédent, mais comme j'utilise CompletableFuture, j'utilise Runnable, et avec la méthode CompletableFuture.allOf, cette fois ça fait du bien et tout se termine. Je t'attendrai. C'est celui que je voulais.
Lorsque vous avez terminé, vous pouvez utiliser la même boucle que la dernière fois, mais lorsque tous les threads sont correctement terminés, vous pouvez utiliser une écriture de type Linq pour récupérer la valeur en tant que lambda et la sortir.
public class CompletableFanOutFanIn {
public void execute() throws InterruptedException, ExecutionException {
ExecutorService executor = Executors.newFixedThreadPool(5);
// ExecutorService executor = Executors.newSingleThreadExecutor();
List<CompletableFuture<String>> results = new ArrayList<>();
for (int i = 0; i < 10; i++) {
CompletableFuture<String> result = runAsync(new RunnableActivity(), executor);
results.add(result);
}
CompletableFuture<Void> cf = CompletableFuture.allOf(results.toArray(new CompletableFuture[10]));
cf.whenComplete((ret, ex) -> {
if(ex == null) {
String msg = results.stream().map(future -> {
try {
return future.get();
} catch (Exception iex) {
return iex.toString();
}
}).collect(Collectors.joining("\n"));
System.out.println("result = " + msg);
} else {
System.err.println(ex);
}
});
}
private CompletableFuture<String> runAsync(RunnableActivity runnable, ExecutorService executor) {
CompletableFuture<String> cf = new CompletableFuture<>();
executor.execute(() -> {
runnable.run();
cf.complete(runnable.getResult());
});
return cf;
}
}
Result
À propos, avec C # Task, Thread.id est partagé à la place, mais Java semble clairement être un thread.
[23] Activity: 0
[22] Activity: 0
[21] Activity: 0
[20] Activity: 0
[24] Activity: 0
[20] Activity: 1
[23] Activity: 1
[22] Activity: 1
[24] Activity: 1
[21] Activity: 1
[20] Activity: 2
:
[20] Activity: 97
[24] Activity: 97
[22] Activity: 97
[20] Activity: 98
[23] Activity: 98
[21] Activity: 98
[24] Activity: 98
[22] Activity: 98
[22] Activity: 99
[22] Done
[24] Activity: 99
[24] Done
[23] Activity: 99
[23] Done
[21] Activity: 99
[21] Done
[20] Activity: 99
[20] Done
result = Thread [20] has done.
Thread [21] has done.
Thread [22] has done.
Thread [23] has done.
Thread [24] has done.
Thread [21] has done.
Thread [20] has done.
Thread [22] has done.
Thread [23] has done.
Thread [24] has done.
Pour être honnête, je ne suis toujours pas confiant dans les programmes concurrents après Java8. Prenez votre temps et apprenez dur. Autour de Pluralsight. Ne vous précipitez pas, étape par étape. Tout d'abord, j'ai pu mettre en œuvre ce que je pensais à l'origine être la première étape, je suis donc heureux aujourd'hui.
Recommended Posts