Lire le fichier volumineux multithread

Question

J'implémente une classe qui devrait recevoir un gros fichier texte. Je veux le diviser en morceaux et chaque morceau doit être tenu par un thread différent qui comptera la fréquence de chaque caractère dans ce morceau. Je m'attends à démarrer plus de threads pour obtenir de meilleures performances, mais il s'avère que les performances sont de plus en plus pauvres. Here`s mon code:

public class Main { 

    public static void main(String[] args) 
    throws IOException, InterruptedException, ExecutionException, ParseException 
    { 

     // save the current run's start time 
     long startTime = System.currentTimeMillis(); 

     // create options 
     Options options = new Options(); 
     options.addOption("t", true, "number of threads to be start"); 

     // variables to hold options 
     int numberOfThreads = 1; 

     // parse options 
     CommandLineParser parser = new DefaultParser(); 
     CommandLine cmd; 
     cmd = parser.parse(options, args); 
     String threadsNumber = cmd.getOptionValue("t"); 
     numberOfThreads = Integer.parseInt(threadsNumber); 

     // read file 
     RandomAccessFile raf = new RandomAccessFile(args[0], "r"); 
     MappedByteBuffer mbb 
      = raf.getChannel().map(FileChannel.MapMode.READ_ONLY, 0, raf.length()); 

     ExecutorService pool = Executors.newFixedThreadPool(numberOfThreads); 
     Set<Future<int[]>> set = new HashSet<Future<int[]>>(); 

     long chunkSize = raf.length()/numberOfThreads; 
     byte[] buffer = new byte[(int) chunkSize]; 

     while(mbb.hasRemaining()) 
     { 
      int remaining = buffer.length; 
      if(mbb.remaining() < remaining) 
      { 
       remaining = mbb.remaining(); 
      } 
      mbb.get(buffer, 0, remaining); 
      String content = new String(buffer, "ISO-8859-1"); 
      @SuppressWarnings("unchecked") 
      Callable<int[]> callable = new FrequenciesCounter(content); 
      Future<int[]> future = pool.submit(callable); 
      set.add(future); 

     } 

     raf.close(); 

     // let`s assume we will use extended ASCII characters only 
     int alphabet = 256; 

     // hold how many times each character is contained in the input file 
     int[] frequencies = new int[alphabet]; 

     // sum the frequencies from each thread 
     for(Future<int[]> future: set) 
     { 
      for(int i = 0; i < alphabet; i++) 
      { 
       frequencies[i] += future.get()[i]; 
      } 
     } 
    } 

} 

//help class for multithreaded frequencies` counting 
class FrequenciesCounter implements Callable 
{ 
    private int[] frequencies = new int[256]; 
    private char[] content; 

    public FrequenciesCounter(String input) 
    { 
     content = input.toCharArray(); 
    } 

    public int[] call() 
    { 
     System.out.println("Thread " + Thread.currentThread().getName() + "start"); 

     for(int i = 0; i < content.length; i++) 
     { 
      frequencies[(int)content[i]]++; 
     } 

     System.out.println("Thread " + Thread.currentThread().getName() + "finished"); 

     return frequencies; 
    } 
} 

Answer 1

Comme suggéré dans les commentaires, vous allez (en général) ne reçoivent pas de meilleures performances lors de la lecture de plusieurs threads. Plutôt vous devriez traiter les morceaux que vous avez lus sur plusieurs threads. Habituellement, le traitement effectue des opérations d'E/S bloquantes (enregistrement dans un autre fichier, enregistrement dans la base de données? Appel HTTP?) Et vos performances s'améliorent si vous traitez plusieurs threads.

Pour le traitement, vous pouvez avoir ExecutorService (avec un nombre raisonnable de threads). utiliser java.util.concurrent.Executors pour obtenir instance de java.util.concurrent.ExecutorService

Ayant ExecutorService exemple, vous pouvez submit vos morceaux pour le traitement. Soumettre des morceaux ne bloquerait pas. ExecutorService commencera à traiter chaque tronçon à thread séparé (les détails dépendent de la configuration de ExecutorService). Vous pouvez soumettre des instances de Runnable ou Callable.

Enfin, après avoir soumis tous les articles, vous devez appeler awaitTermination à votre ExecutorService. Il attendra que le traitement de tous les éléments soumis soit terminé. Après le retour de awaitTermination, vous devez appeler shutdownNow() pour abandonner le traitement (sinon, il risque de se bloquer indéfiniment et de traiter une tâche indésirable).

Answer 2

Votre programme est presque certainement limité par la vitesse de lecture à partir du disque. L'utilisation de plusieurs threads n'aide pas car la limite est une limite matérielle sur la vitesse à laquelle les informations peuvent être transférées à partir du disque. En outre, l'utilisation de RandomAccessFile et d'un tampon suivant entraîne probablement un léger ralentissement, puisque vous déplacez les données en mémoire après les avoir lues mais avant de les traiter, plutôt que de simplement les traiter en place. Vous feriez mieux de ne pas utiliser un tampon intermédiaire.

Vous pouvez obtenir une légère accélération en lisant le fichier directement dans les tampons finaux et en envoyant ces tampons à traiter par les threads au fur et à mesure qu'ils sont remplis, plutôt que d'attendre la lecture complète du fichier avant le traitement. Cependant, la plupart du temps sera toujours utilisé par le disque lu, donc toute accélération serait probablement minime.