Warteschlangen verwalten Daten nach dem First-In-First-Out-Prinzip. Wenn Sie versuchen, ein Element zu einer vollständigen Blockierungswarteschlange hinzuzufügen oder ein Element aus einer leeren Blockierungswarteschlange zu entfernen, wird der Thread blockiert. Blockierungswarteschlangen sind ein nützliches Werkzeug, wenn mehrere Threads zusammenarbeiten. Der Arbeitsthread kann Zwischenergebnisse regelmäßig in der Blockierungswarteschlange speichern. Andere Arbeitsthreads nehmen die Zwischenergebnisse heraus und ändern sie in Zukunft. Die Warteschlange gleicht die Last automatisch aus. Wenn der erste Thread-Satz langsamer läuft als der zweite Thread-Satz, blockiert der zweite Thread-Satz, während er auf Ergebnisse wartet. Wenn der erste Thread-Satz schnell läuft, wird darauf gewartet, dass der zweite Thread-Satz aufholt.
Das folgende Programm zeigt, wie Blockierungswarteschlangen zur Steuerung von Thread-Gruppen verwendet werden. Das Programm durchsucht alle Dateien in einem Verzeichnis und allen seinen Unterverzeichnissen und gibt eine Liste der Dateien aus, die das angegebene Schlüsselwort enthalten.
Das Paket java.util.concurrent bietet vier Varianten von Blockierungswarteschlangen: LinkedBlockingQueue, ArrayBlockingQueue, PriorityBlockingQueue und DelayQueue. Wir verwenden ArrayBlockingQueue. ArrayBlockingQueue erfordert beim Aufbau eine bestimmte Kapazität und kann optional Fairness erfordern. Wenn der Fairness-Parameter gesetzt ist, wird der Thread mit der längsten Wartezeit zuerst verarbeitet. Im Allgemeinen kostet Fairness Leistung, also nutzen Sie sie nur, wenn Sie sie wirklich brauchen.
Der Producer-Thread zählt alle Dateien in allen Unterverzeichnissen auf und stellt sie in eine Blockierungswarteschlange. Dieser Vorgang ist schnell und wenn die Warteschlange nicht begrenzt ist, enthält sie bald Dateien, die nicht gefunden wurden.
Wir haben auch eine große Anzahl von Suchthreads gestartet. Jeder Suchthread nimmt eine Datei aus der Warteschlange, öffnet sie, gibt alle Zeilen aus, die das Schlüsselwort enthalten, und nimmt dann die nächste Datei. Wir verwenden einen kleinen Trick, um den Thread nach getaner Arbeit zu beenden. Um den Abschluss zu signalisieren, stellt der Aufzählungsthread ein virtuelles Objekt in die Warteschlange. (Dies ähnelt dem Ablegen einer virtuellen Tasche auf einem Gepäckband mit der Aufschrift „Letzte Tasche“.) Wenn der Suchthread das virtuelle Objekt abruft, legt er es zurück und wird beendet.
Beachten Sie, dass hier keine explizite Thread-Synchronisierung erforderlich ist. In diesem Programm verwenden wir die Warteschlangendatenstruktur als Synchronisationsmechanismus.
Kopieren Sie den Codecode wie folgt:
java.io.* importieren;
import java.util.*;
import java.util.concurrent.*;
öffentliche Klasse BlockingQueueTest
{
public static void main(String[] args)
{
Scanner in = neuer Scanner(System.in);
System.out.print("Basisverzeichnis eingeben (z. B. /usr/local/jdk1.6.0/src):");
String-Verzeichnis = in.nextLine();
System.out.print("Schlüsselwort eingeben (z. B. volatile):");
String-Schlüsselwort = in.nextLine();
final int FILE_QUEUE_SIZE = 10;
final int SEARCH_THREADS = 100;
BlockingQueue<File> queue = new ArrayBlockingQueue<File>(FILE_QUEUE_SIZE);
FileEnumerationTask enumerator = new FileEnumerationTask(queue, new File(directory));
neuer Thread(enumerator).start();
for (int i = 1; i <= SEARCH_THREADS; i++)
new Thread(new SearchTask(queue, keyword)).start();
}
}
/**
* Diese Aufgabe zählt alle Dateien in einem Verzeichnis und seinen Unterverzeichnissen auf.
*/
Die Klasse FileEnumerationTask implementiert Runnable
{
/**
* Konstruiert eine FileEnumerationTask.
* @param queue Die Blockierungswarteschlange, zu der die aufgelisteten Dateien hinzugefügt werden
* @param StartingDirectory das Verzeichnis, in dem die Aufzählung gestartet werden soll
*/
public FileEnumerationTask(BlockingQueue<File> queue, File StartingDirectory)
{
this.queue = Warteschlange;
this.startingDirectory = StartingDirectory;
}
public void run()
{
versuchen
{
enumerate(startingDirectory);
queue.put(DUMMY);
}
Catch (InterruptedException e)
{
}
}
/**
* Listet rekursiv alle Dateien in einem bestimmten Verzeichnis und seinen Unterverzeichnissen auf
* @param-Verzeichnis das Verzeichnis, in dem gestartet werden soll
*/
public void enumerate(Dateiverzeichnis) löst eine InterruptedException aus
{
Datei[] Dateien = Verzeichnis.listFiles();
for (Datei Datei: Dateien)
{
if (file.isDirectory()) enumerate(file);
sonst queue.put(file);
}
}
öffentliche statische Datei DUMMY = new File("");
private BlockingQueue<File>-Warteschlange;
privates Dateistartverzeichnis;
}
/**
* Diese Aufgabe durchsucht Dateien nach einem bestimmten Schlüsselwort.
*/
Die Klasse SearchTask implementiert Runnable
{
/**
* Konstruiert eine SearchTask.
* @param queue die Warteschlange, aus der Dateien entnommen werden sollen
* @param keyword das zu suchende Schlüsselwort
*/
public SearchTask(BlockingQueue<File> queue, String keyword)
{
this.queue = Warteschlange;
this.keyword = Schlüsselwort;
}
public void run()
{
versuchen
{
boolean done = false;
while (!done)
{
Datei file = queue.take();
if (file == FileEnumerationTask.DUMMY)
{
queue.put(file);
erledigt = wahr;
}
sonst suche(Datei);
}
}
Catch (IOException e)
{
e.printStackTrace();
}
Catch (InterruptedException e)
{
}
}
/**
* Durchsucht eine Datei nach einem bestimmten Schlüsselwort und gibt alle passenden Zeilen aus.
* @param file die zu durchsuchende Datei
*/
public void search(File file) löst eine IOException aus
{
Scanner in = new Scanner(new FileInputStream(file));
int lineNumber = 0;
while (in.hasNextLine())
{
Zeilennummer++;
String line = in.nextLine().trim();
if (line.contains(keyword)) System.out.printf("%s:%d %s%n", file.getPath(), lineNumber, line);
}
in.close();
}
private BlockingQueue<File>-Warteschlange;
privates String-Schlüsselwort;
}