Comment puis-je paralléliser le mappage d'une liste?

Question

C'est le problème que j'ai: J'ai une grande séquence de certains objets (List<SomeClass>), et je veux effectuer une opération sur tous les éléments de la liste et obtenir une nouvelle séquence (List<SomeOtherClass>).

Comme:

List<SomeOtherClass> list = new ArrayList<SomeOtherClass>(); 
for(SomeClass sc : originalList) 
    list.add(someOperation(sc)); 

Depuis l'opération someOperation n'a pas d'effets secondaires, et la liste est assez grand, je veux cette opération de cartographie à parallélisé.

Quelle sera la meilleure façon de faire cela en Java?

Answer 1

Si vous êtes sûr qu'il n'y a pas d'effet secondaire, utilisez simplement le filetage. Bien sûr, vous pouvez utiliser le filetage lorsqu'il y a un effet secondaire. Mais dans ce cas, vous devrez utiliser des mécanismes de verrouillage et de synchronisation.

Un échantillon de fil très simple de l'excellent livre de Bruce Eckel, Penser en Java:

public class SimpleThread extends Thread { 
    private int countDown = 5; 
    private static int threadCount = 0; 
    public SimpleThread() { 
    super("" + ++threadCount); // Store the thread name 
    start(); 
    } 
    public String toString() { 
    return "#" + getName() + ": " + countDown; 
    } 
    public void run() { 
    while(true) { 
     System.out.println(this); 
     if(--countDown == 0) return; 
    } 
    } 
    public static void main(String[] args) { 
    for(int i = 0; i < 5; i++) 
     new SimpleThread(); 
    } 
} ///:~ 

Answer 2

Une implémentation possible peut utiliser le cadre Executor (exemple inclus).

Answer 3

Utilisez le filetage et partitionnez votre travail à l'aide de sous-listes.

Answer 4

Divisez la liste d'entrée et utilisez la tâche FutureTask, puis fusionner les résultats

Answer 5

Voici comment je le fais habituellement il:

// Somewhere define a static final int NUM_THREADS that is appropriate. 

ExecutorService exec = Executors.newFixedThreadPool(NUM_THREADS); 
// There are other options: look at what the Executors class has to offer. 

List<SomeOtherClass> list = new ArrayList<SomeOtherClass>(); 

List<Future<SomeOtherClass>> list = new ArrayList<Future<SomeOtherClass>>(); 

for(SomeClass sc : originalList) 
    futures.add(submit(new someOperation(sc))); 

for(Future<SomeOtherClass> future : futures) 
    list.add(future.get()); // Again, see the docs, you can also set a timeout. 

exec.shutdown(); // Important. Otherwise the threads stay alive. 

someOperation est alors défini comme un appelable

class someOperation extends Callable<SomeOtherClass> { 

    private SomeClass input; 

    public someOperation(SomeClass input){ 
     this.input = input; 
    } 

    public SomeOtherClass call(){ 
     // Do your operation on 'input' here 
    } 
} 

Note: Je n'avais aucun try - catch blocs ici, mais vous devrez en avoir. shutdown devrait être dans le bloc finally. Je ne me souviens pas de ce qui jette quel ATM, votre IDE devrait vous aider avec ça.

Answer 6

Le cadre jsr166y de Doug Lea comprend le cadre Fork/Join, parfait pour ce genre de choses. Il est conçu pour la décomposition parallèle des tâches intensives du CPU comme celle-ci, et il est très efficace. La façon la plus facile de l'utiliser est probablement la nouvelle Scala 2.9 parallel collections car elle complète complètement son utilisation (pour obtenir une collection parallèle dans Scala, il suffit d'ajouter .par).

Les deux solutions ont de nombreux tutoriels et exemples disponibles via Google.