Java. Consommateur - Producteur avec BlockingQueue. Outil de recherche

Question

J'essayais d'implémenter un problème de Consumer-Producer avec BlockingQueue. Pour le faire avec un certain but, j'ai décidé d'écrire l'outil de recherche de fichiers.

J'ai décidé que le mécanisme de recherche fonctionnait récursivement, et chaque nouveau répertoire va avoir un nouveau pool de threads pour accélérer la recherche. Mon problème est que je n'ai aucune idée de la façon dont je peux mettre en œuvre un mécanisme qui arrête d'imprimer des threads (consommateurs) à la fin - lors de la recherche de threads faire le travail. J'essayais de le faire avec des idées comme POISON PILLS, mais ça ne marche pas bien (les threads s'arrêtent avant d'imprimer les résultats). Des idées comment puis-je faire cela?

Voici un code:

mécanisme de recherche:

public class SearchingAlgorithm implements Runnable { 

private final File file; 
private BlockingQueue<File> queue; 
private ExecutorService executor; 

public SearchingAlgorithm(File fileName, BlockingQueue<File> queue) { 
    this.file = fileName; 
    this.queue = queue; 
    this.executor = Executors.newWorkStealingPool(); 
} 

@Override 
public void run() { 
    try { 
     searchDeep(); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 
} 

private void searchDeep() throws InterruptedException { 
    File[] files = file.listFiles(); 
    if (files != null) { 
     for (File fil : files) { 
      if (fil.isDirectory()) { 
       executor.submit(new SearchingAlgorithm(fil, this.queue)); 
      } else { 
       this.queue.add(fil); 
      } 
     } 
    } 
} 

}

Imprimante:

public class ContainingCheckAlgorithm implements Runnable { 

private BlockingQueue<File> queue; 
// private ExecutorService executor; 
private String keyWord; 

public ContainingCheckAlgorithm(BlockingQueue<File> queue, String keyWord) { 
    this.queue = queue; 
    this.keyWord = keyWord; 
    // executor = Executors.newFixedThreadPool(2); 
} 

@Override 
public void run() { 
    try { 
     printFile(); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 

} 

private void printFile() throws InterruptedException { 
    while (true) { 
     File takeFile = queue.take(); 
     String fileName = takeFile.getAbsolutePath() 
       .toLowerCase(); 
     boolean isContainingKeyWord = fileName.contains(keyWord.toLowerCase()); 

     if (isContainingKeyWord) { 
      System.out.println(takeFile.getAbsolutePath()); 
     } 
    } 
} 

}

classe de test principal:

public class MainClass { 

public static void main(String[] args) throws InterruptedException { 
    ExecutorService executor = Executors.newFixedThreadPool(2); 
    BlockingQueue<File> queue = new LinkedBlockingQueue<>(); 

    File fileName = new File("C:/"); 

    SearchingAlgorithm sa = new SearchingAlgorithm(fileName, queue); 
    executor.submit(sa); 

    ContainingCheckAlgorithm ca = new ContainingCheckAlgorithm(queue, "Slipknot"); 
    executor.submit(ca); 

    executor.shutdown(); 
} 

}

Answer 1

Sépare l'ensemble des travaux en 2 étapes. Lors de la première étape, le travail de SearchingAlgorithm et ContainingCheckAlgorithm attendent de nouveaux travaux si la file d'attente est vide. À la deuxième étape, toutes les instances de SearchingAlgorithm ont terminé et ContainingCheckAlgorithm se ferme si la file d'attente est vide. Pour découvrir si la file d'attente est vide, ContainingCheckAlgorithm utilise queue.poll (timeout) au lieu de queue.take().

Et vous n'avez pas besoin de créer un nouveau pool de threads pour chaque SearchingAlgorithm.

Answer 2

Comme u dit, j'essaie de le faire de cette façon:

Recherche algorithme part du pool de threads avec d'autres insatances de searchingAlgorithm.

RECHERCHE:

public class SearchingAlgorithm implements Runnable { 

private final File file; 
private BlockingQueue<File> queue; 
private ExecutorService executor; 

public SearchingAlgorithm(File fileName, BlockingQueue<File> queue, ExecutorService executor) { 
    this.file = fileName; 
    this.queue = queue; 
    this.executor = executor; 
} 

@Override 
public void run() { 
    try { 
     searchDeep(); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 
} 

private void searchDeep() throws InterruptedException { 
    File[] files = file.listFiles(); 
    if (files != null) { 
     for (File fil : files) { 
      if (fil.isDirectory()) { 
       executor.submit(new SearchingAlgorithm(fil, this.queue, executor)); 
      } else { 
       this.queue.add(fil); 
      } 
     } 
    } 
} 

maintenant besoin de partager ContainingCheckAlgorith CountDownLatch avec classe principale, parce que je besoin d'un mécanisme pour fermer du pool de threads dans la classe principale. En outre, il utilise pool (timeout) comme vous l'avez dit, et mes threads finissent finalement leur travail.

VÉRIFIER

public class ContainingCheckAlgorithm implements Runnable { 

private BlockingQueue<File> queue; 
private String keyWord; 
private CountDownLatch latch; 

public ContainingCheckAlgorithm(BlockingQueue<File> queue, String keyWord, CountDownLatch latch) { 
    this.queue = queue; 
    this.keyWord = keyWord; 
    this.latch = latch; 
} 

@Override 
public void run() { 
    try { 
     printFile(); 
     latch.countDown(); 
    } catch (InterruptedException e) { 
     e.printStackTrace(); 
    } 

} 

private void printFile() throws InterruptedException { 
    File takeFile; 
    while ((takeFile = queue.poll(1, TimeUnit.SECONDS)) != null) { 
     String fileName = takeFile.getName() 
       .toLowerCase(); 
     boolean isContainingKeyWord = fileName.contains(keyWord.toLowerCase()); 

     if (isContainingKeyWord) { 
      System.out.println(takeFile.getAbsolutePath()); 
     } 
    } 
} 

PRINCIPAL:

public class MainClass { 

public static void main(String[] args) throws InterruptedException { 
    ExecutorService executor = Executors.newCachedThreadPool(); 
    BlockingQueue<File> queue = new LinkedBlockingQueue<>(); 
    CountDownLatch latch = new CountDownLatch(1); 

    File fileName = new File("C:/"); 

    SearchingAlgorithm sa = new SearchingAlgorithm(fileName, queue, executor); 
    executor.submit(sa); 

    ContainingCheckAlgorithm ca = new ContainingCheckAlgorithm(queue, "Slipknot", latch); 
    executor.submit(ca); 

    latch.await(); 
    executor.shutdown(); 
} 

Il fait un peu bizarre, mais je me demande si:

• Plus de 1 fil irait comme ContainingCheckAlgorithm ? SearchingAlgorithm rechercherait un fichier plus de 1 seconde, et ContainingCheckAlgorithm terminerait-il son travail? Évidemment, je peux changer le timeout à 2 secondes, et plus, mais nous essayons toujours d'optimiser nos programmes.