So erstellen Sie mehrere Threads für jedes Anforderungselement

9

Ich versuche, den folgenden Code mithilfe von Multithreading auf Auftragsebene zu verarbeiten.

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

Aktuelle sequentielle Ausführung:

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

Ich habe versucht mit:

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

Aber es ändert die Regeln für jede (Regel -> {}} Reihenfolge.

Zum Beispiel:
Eingabe:

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

Erwartete Ausgabe:

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

Tatsächliche Ausgabe mit parallelStream():

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

Die Reihenfolge der Bestellungen stört mich nicht , aber die Reihenfolge der Regeln stört mich . Bestellungen können in beliebiger Reihenfolge verarbeitet werden, Regeln sollten jedoch für jede Bestellung in derselben Reihenfolge ausgeführt werden.

Bitte helfen Sie.

Mayank Bisht
quelle

Antworten:

4

Sie können verwenden:

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

ForEachOrdered garantiert, dass die Reihenfolge des Streams beibehalten wird.

Also als Referenz:

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

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

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

        });

Hinweis: Während wir dies tun können, sollte die Leistung genau beobachtet werden, da Parellelismus und Ordnung sich nicht sehr gut heiraten!

Ausgabe

 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
quelle
Danke für die Antwort. forEachOrdered garantiert die Reihenfolge des Streams, verlangsamt dann aber auch die Leistung. Ich habe es versucht und die Anwendung benötigt Zeit, ähnlich wie bei der sequentiellen Verarbeitung. stream (). parallel & forEachOrdered ergänzen sich nicht.
Mayank Bisht
Ja, ich bin damit einverstanden, dass wir vorher eine vollständige Latenzanalyse durchführen müssen.
Pramod S. Nikam
Ja, ich bekomme die gleiche Leistung, es gibt keine Verbesserung.
Mayank Bisht
1
Folgen Sie diesem Thread genau, um eine bessere Lösung dafür zu erhalten.
Pramod S. Nikam
Kann ich mit ExecutorService eine Parallelverarbeitung erreichen?
Mayank Bisht
1

Sie fügen gleichzeitig Elemente finalListaus verschiedenen Threads hinzu. Dies führt zu Mischergebnissen beim Anwenden von Regeln auf verschiedene Aufträge (Regeln werden nicht nach ihren Aufträgen gruppiert).

Sie können das Problem beheben, indem Sie für jede Liste eine temporäre Liste erstellen orderund dann alle temporären Listen in a synchron zusammenführen finalList.

So können Sie es mit der Stream-API (Java 9+) machen:

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

Hinweis: Collectors.flatMapping()Wird hier verwendet, anstatt flatMapdas Flat Mapping während der Stream-Erfassung einfach synchron auszuführen.


Java 8 analog:

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
quelle
Danke für die Antwort. Ich habe Ihren Ansatz
ausprobiert und erhalte
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, dies bedeutet, dass dies beanMapper.getBean(rule) .applyRule(createTemplate.apply(getMetaData.apply(rule), command), order)keine reine Funktion ist und daher nicht parallel verwendet werden kann. Versuchen Sie, alle Nebenwirkungen zu entfernen. ConcurrentModificationExceptionStack-Trace kann helfen, sie zu finden.
Bananon
0

Ob das funktioniert?

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

Ausgabe

 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
quelle
0

Die Reihenfolge der Bestellungen kann beliebig sein, aber die Reihenfolge der Regeln sollte sich nicht ändern. Auch für eine bestimmte Bestellung sollte Regel in einer Gruppe kommen.

Wenn dies der Fall ist, gibt es keinen Raum für tatsächliche Parallelität.

Wann

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

und

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

sind die einzigen gültigen Läufe für 2 Bestellungen und 2 Regeln,
und

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

wird als ungültig angesehen, das heißt nicht Parallelität, sondern Randomisierung von orders, vermutlich ohne Gewinn. Wenn Sie sich "langweilen" order1, immer an erster Stelle zu stehen, können Sie die Liste mischen, aber das ist alles:

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

Es ist nicht einmal Streaming erforderlich, nur zwei verschachtelte Schleifen. 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


Ursprüngliche Antwort

Aber es ändert die Regeln für jede (Regel -> {}} Reihenfolge.

Nein, tut es nicht. Dasorder s kann sich überlappen, aber die Reihenfolge von rules für jede Reihenfolge wird beibehalten. Warum sollte ein Nicht-Paralleler forEachetwas anderes tun?

Beispielcode:

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
Beispielausgabe:

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

Die Reihenfolge von order s ist gemischt, aber die rules sind immer 1-2-3. Ich denke, Ihre Ausgabe hat die Paarungen einfach versteckt (tatsächlich haben Sie nicht gezeigt, wie sie generiert wurden).

Natürlich kann es mit einigen Verzögerungen verlängert werden, also Verarbeitung von order s tatsächlich überschneidet:

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
Beispielausgabe:

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

Dies kann etwas sein, was Sie gesehen haben, nur ohne das orderxTeil. Mit dem ordersichtbaren s kann verfolgt werden, dass rules immer wieder als 1-2-3 pro kommtorder . Außerdem enthielt Ihre Beispielliste order1zweimal, was sicherlich nicht dazu beitrug, zu sehen, was geschah.

tevemadar
quelle
Danke für die Antwort. Die obige Ausgabe ist möglicherweise für weniger Bestellungen korrekt. Wenn Sie jedoch die Bestellungen erhöhen, erhalten Sie eine andere Ausgabe. Zum Beispiel (order4-rule1 order4-rule2 order4-rule1) (order1-rule1 order1-rule2) (order3-rule1 order3-rule2) (order4-rule1 order4-rule2 order4-rule1 order4-rule2).
Mayank Bisht
Die Reihenfolge der Bestellungen kann beliebig sein, aber die Reihenfolge der Regeln sollte sich nicht ändern. Auch für eine bestimmte Bestellung sollte Regel in einer Gruppe kommen. Zum Beispiel. (order1-rule 1 order1-rule2 order1-rule3) und nicht (order1-rule1 order2-rule1 order1-rule2 order1-rule3).)
mayank bisht
@mayankbisht Ich denke, diese Einschränkungen erlauben einfach keine parallele Verarbeitung. Siehe aktualisierte Antwort (ich habe den neuen Teil am Anfang geschrieben).
Tevemadar
Ja, das verstehe ich und deshalb habe ich diese Frage hier gepostet. Ich dachte, vielleicht gibt es einen anderen Weg, dies zu tun, oder vielleicht können wir das Algo ändern
Mayank Bisht
@mayankbisht Sie könnten beschreiben, warum orders sich nicht überlappen können (sind diese rules möglicherweise zustandsbehaftet und existieren in einer begrenzten Anzahl von Kopien, vielleicht nur einer einzigen?). Aber im Allgemeinen gibt es keine Parallelität ohne parallele Dinge, das ist schließlich der springende Punkt der Parallelität.
Tevemadar
0

Wenn es Ihnen nichts ausmacht, die Bibliothek eines Drittanbieters auszuprobieren. Hier ist ein Beispiel mit meiner Bibliothek: abacus-util

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

Und Sie können sogar die Thread-Nummer angeben:

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

Die Reihenfolge von rule wird eingehalten.

Übrigens, da es sich um einen parallelen Stream handelt, wird der Code ...finalList.add(...höchstwahrscheinlich nicht funktionieren. Ich denke, es ist besser, das Ergebnis zu sammeln, um es aufzulisten:

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

Dies ist auch dann möglich, wenn Sie die Reihenfolge orderspäter aus irgendeinem Grund beibehalten möchten :

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