Comment créer plusieurs threads pour chaque élément de demande

9

J'essaie de traiter le code ci-dessous en utilisant le multithreading au niveau de la commande.

List<String> orders = Arrays.asList("order1", "order2", 
                   "order3", "order4", "order1");

Exécution séquentielle actuelle:

orders.stream().forEach(order -> {
    rules.forEach(rule -> {
        finalList.add(beanMapper.getBean(rule)
                .applyRule(createTemplate.apply(getMetaData.apply(rule), command),
                           order));
    });
});

J'ai essayé d'utiliser:

orders.parallelStream().forEach(order -> {}} // code snippet.

Mais cela modifie l'ordre rules.forEach (rule -> {}} .

Par exemple:
Entrée:

 List<String> orders = Arrays.asList("order1", "order2", 
                         "order3", "order4", "order1");
 List<String> rules = Arrays.asList("rule1", "rule2", "rule3");

Production attendue:

order1 with rule1, rule2, rule3
order2 with rule1, rule2, rule3

Sortie réelle avec parallelStream():

order1 with rule3, rule1, rule2
order1 with rule2, rule1, rule3

Je ne m'inquiète pas de l'ordre des commandes , mais je m'inquiète de l'ordre des règles . Les commandes peuvent être traitées dans n'importe quel ordre, mais les règles doivent s'exécuter dans le même ordre pour chaque ordre.

Veuillez aider.

mayank bisht
la source

Réponses:

4

Vous pouvez utiliser :

orders.stream().parallel().forEachOrdered(// Your rules logic goes here. )

ForEachOrdered garantit le maintien de l'ordre du Stream.

Donc pour votre référence:

orders.stream().parallel().forEachOrdered( order -> {

            rules.stream().parallel().forEachOrdered ( rule -> {

                 System.out.println( " Order : " + order + " rule :" + rule);
            });

        });

Remarque: Bien que nous puissions le faire, les performances doivent être surveillées de près car le parellélisme et l'ordre ne se marient pas très bien!

Production

 Order : order1 rule :rule1
 Order : order1 rule :rule2
 Order : order1 rule :rule3
 Order : order2 rule :rule1
 Order : order2 rule :rule2
 Order : order2 rule :rule3
 Order : order3 rule :rule1
 Order : order3 rule :rule2
 Order : order3 rule :rule3
 Order : order4 rule :rule1
 Order : order4 rule :rule2
 Order : order4 rule :rule3
 Order : order1 rule :rule1
 Order : order1 rule :rule2
 Order : order1 rule :rule3
Pramod S. Nikam
la source
Merci d'avoir répondu. forEachOrdered garantit l'ordre du flux, mais ralentit également les performances. Je l'ai essayé et l'application prend un temps similaire au traitement séquentiel. stream (). parallel & forEachOrdered ne se complimentent pas mutuellement.
Mayank Bisht,
Oui, je suis d'accord que nous devons faire une analyse complète de la latence avant de le faire.
Pramod S. Nikam
Oui, j'obtiens les mêmes performances en utilisant cela, il n'y a pas d'amélioration.
mayank bisht
1
Suivre de près ce fil pour obtenir une meilleure solution pour ce faire.
Pramod S. Nikam
Puis-je réaliser un traitement parallèle en utilisant ExecutorService?
Mayank Bisht
1

Vous ajoutez des éléments aux finalListdifférents threads en même temps. Cela provoque un mélange des résultats de l'application des règles à différents ordres (les règles ne sont pas regroupées par leurs ordres).

Vous pouvez le corriger en créant une liste temporaire pour chacun order, puis en fusionnant de manière synchrone toutes les listes temporaires dans un fichier finalList.

Voici comment vous pouvez le faire en utilisant Stream-API (Java 9+):

List<AppliedRule> finalList = orders.parallelStream().map(order ->
        rules.stream().map(rule -> applyRule(order, rule)).collect(Collectors.toList())
).collect(Collectors.flatMapping(Collection::stream, Collectors.toList()));

Remarque: Collectors.flatMapping()est utilisé ici au lieu de simple flatMappour exécuter un mappage plat de manière synchrone pendant la collecte de flux.


Analogue Java 8:

List<AppliedRule> finalList = orders.parallelStream().map(order ->
        rules.stream().map(rule -> applyRule(order, rule)).collect(Collectors.toList())
).collect(Collectors.toList())
        .stream()
        .flatMap(Collection::stream)
        .collect(Collectors.toList());
Bananon
la source
Merci d'avoir répondu. J'ai essayé votre approche et je reçois java.util.ConcurrentModificationException: null
mayank bisht
finalList = orders.parallelStream () .map (order -> rules.stream () .map (rule -> beanMapper.getBean (rule) .applyRule (createTemplate.apply (getMetaData.apply (rule), command), order))) .collect (Collectors.toList ())). collect (Collectors.toList ()). stream (). flatMap (Collection :: stream) .collect (Collectors.toList ());
Mayank Bisht
@mayankbisht, cela signifie que ce beanMapper.getBean(rule) .applyRule(createTemplate.apply(getMetaData.apply(rule), command), order)n'est pas une fonction pure, donc elle ne peut pas être utilisée en parallèle. Essayez d'en supprimer tous les effets secondaires; ConcurrentModificationExceptionla trace de la pile peut aider à les localiser.
Bananon
0

Est-ce que ça va marcher?

final int rulesSize = rules.size();
AtomicInteger atomicInteger = new AtomicInteger(0);
orders.stream().parallel().forEach(order -> {
    IntStream.range(0, rulesSize).parallel().forEach( i -> {
        synchronized (atomicInteger) {
            System.out.println(" Order : " + order + " rule :" + rules.get(atomicInteger.getAndIncrement() % rulesSize));
        }
    });
});

Production

 Order : order1 rule :rule1
 Order : order4 rule :rule2
 Order : order1 rule :rule3
 Order : order3 rule :rule1
 Order : order3 rule :rule2
 Order : order3 rule :rule3
 Order : order2 rule :rule1
 Order : order2 rule :rule2
 Order : order2 rule :rule3
 Order : order1 rule :rule1
 Order : order1 rule :rule2
 Order : order4 rule :rule3
 Order : order1 rule :rule1
 Order : order4 rule :rule2
 Order : order1 rule :rule3
BHAWANI SINGH
la source
0

L'ordre des commandes peut être n'importe quoi, mais l'ordre des règles ne doit pas changer. Aussi pour une règle d'ordre particulière devrait venir dans un groupe.

Si tel est le cas, il n'y a pas de place pour un parallélisme réel.

Quand

order1-rule1
order1-rule2
order2-rule1
order2-rule2

et

order2-rule1
order2-rule2
order1-rule1
order1-rule2

sont les seules séries valides pour 2 commandes et 2 règles,
et

order1-rule1
order2-rule1
order1-rule2
order2-rule2

est considéré comme invalide, ce n'est pas du parallélisme, juste une randomisation de orders, sans doute sans gain. Si vous "vous ennuyez" en order1venant tout le temps en premier, vous pouvez mélanger la liste, mais c'est tout:

public static void main (String[] args) throws java.lang.Exception
{
    List<String> orders = Arrays.asList("order1", "order2", "order3", "order4");
    List<String> rules = Arrays.asList("rule1", "rule2", "rule3");
    Collections.shuffle(orders);
    orders.forEach(order->{
        rules.forEach(rule->{
            System.out.println(order+"-"+rule);
        });
    });
}

Même le streaming n'est pas nécessaire, juste deux boucles imbriquées. Test: https://ideone.com/qI3dqd

order2-rule1
order2-rule2
order2-rule3
order4-rule1
order4-rule2
order4-rule3
order1-rule1
order1-rule2
order1-rule3
order3-rule1
order3-rule2
order3-rule3


Réponse originale

Mais cela modifie l'ordre rules.forEach (rule -> {}}.

Non. Les orders peuvent se chevaucher, mais l'ordre des rules pour chaque commande est conservé. Pourquoi un non-parallèle forEachferait-il autre chose?

Exemple de code:

public static void main (String[] args) throws java.lang.Exception
{
    List<String> orders = Arrays.asList("order1", "order2", "order3", "order4");
    List<String> rules = Arrays.asList("rule1", "rule2", "rule3");
    orders.stream().parallel().forEach(order->{
        rules.forEach(rule->{
            System.out.println(order+"-"+rule);
        });
    });
}

Test: https://ideone.com/95Cybg
Exemple de sortie:

order2-rule1
order2-rule2
order2-rule3
order1-rule1
order1-rule2
order1-rule3
order4-rule1
order4-rule2
order4-rule3
order3-rule1
order3-rule2
order3-rule3

L'ordre des orders est mixte, mais les rules sont toujours 1-2-3. Je pense que votre sortie a simplement caché les appariements (en fait, vous n'avez pas montré comment elle a été générée).

Bien sûr, il peut être étendu avec quelques retards, donc le traitement des orders se chevauchera en fait:

public static void delay(){
    try{
        Thread.sleep(ThreadLocalRandom.current().nextInt(100,300));
    }catch(Exception ex){}
}

public static void main (String[] args) throws java.lang.Exception
{
    List<String> orders = Arrays.asList("order1", "order2", "order3", "order4");
    List<String> rules = Arrays.asList("rule1", "rule2", "rule3");
    orders.stream().parallel().forEach(order->{
        rules.forEach(rule->{
            delay();
            System.out.println(order+"-"+rule);
        });
    });
}

Test: https://ideone.com/cSFaqS
Exemple de sortie:

order3-rule1
order2-rule1
order2-rule2
order3-rule2
order3-rule3
order2-rule3
order1-rule1
order4-rule1
order1-rule2
order4-rule2
order4-rule3
order1-rule3

Cela peut être quelque chose que vous avez vu, juste sans la orderxpartie. Avec le orders visible, il peut être suivi que rules continue à venir comme 1-2-3, parorder . De plus, votre liste d'exemples contenait order1deux fois, ce qui n'a certainement pas aidé à voir ce qui se passait.

tevemadar
la source
Merci d'avoir répondu. La sortie ci-dessus peut être correcte pour moins de commandes. Mais si vous augmentez les commandes, vous obtiendrez une sortie différente. Par exemple (order4-rule1 order4-rule2 order4-rule1) (order1-rule1 order1-rule2) (order3-rule1 order3-rule2) (order4-rule1 order4-rule2 order4-rule1 order4-rule2).
Mayank Bisht
L'ordre des commandes peut être n'importe quoi, mais l'ordre des règles ne doit pas changer. Aussi pour une règle d'ordre particulière devrait venir dans un groupe. Par exemple. (ordre1- règle 1 ordre1-règle2 ordre1-règle3) et non (ordre1-règle1 ordre2-règle1 ordre1-règle2 ordre1-règle3).)
mayank bisht
@mayankbisht Je pense que ces restrictions ne permettent tout simplement pas le traitement parallèle. Voir la réponse mise à jour (j'ai écrit la nouvelle partie au début).
tevemadar
Ouais, je comprends cela, et c'est pourquoi j'ai posté cette question ici. Je pensais qu'il y aurait peut-être une autre façon de le faire, ou peut-être que nous pourrions changer l'algo
mayank bisht
@mayankbisht, vous pourriez décrire pourquoi les orders ne peuvent pas se chevaucher (ceux-ci sont-ils rulepeut-être avec état et existent-ils en un nombre limité d'exemplaires, peut-être un seul?). Mais généralement, il n'y a pas de parallélisme sans que les choses tournent en parallèle, c'est tout l'intérêt du parallélisme après tout.
tevemadar
0

Si cela ne vous dérange pas d'essayer une bibliothèque tierce. Voici un exemple avec ma bibliothèque: abacus-util

StreamEx.of(orders).parallelStream().forEach(order -> {}}

Et vous pouvez même spécifier le numéro de fil:

StreamEx.of(orders).parallelStream(maxThreadNum).forEach(order -> {}}

L'ordre de rulesera conservé.

Soit dit en passant, puisqu'il est en flux parallèle, le morceau de code ...finalList.add(...ne fonctionnera probablement pas. Je pense qu'il vaut mieux collecter le résultat pour lister:

StreamEx.of(orders).parallelStream().map/flatMap(order -> {...}}.toList()

il est également possible même si vous souhaitez conserver l'ordre de orderpour une raison ultérieure:

StreamEx.of(orders).indexed().parallelStream()
      .map/flatMap(order -> {...}}.sortedBy(...index).toList()
user_3380739
la source