IndexingService.java shows an example of stopping a single producer single consumer pair with a poison pill.
The object private static final File POISON = new File(""); is the poison pill the producer and consumer agreed on. Once CrawlerThread get interrupted while in crawl(root); code block, the CrawlerThread stubbornly put the the poison pill queue.put(POISON); to the queue before breaks the loop and exits. The consumer thread IndexerThread, takes the poison pill object from the queue as any other queue element. Once saw the if (file == POISON) match, it breaks the while loop and exits.
The main thread started both CrawlerThread and IndexerThread by calling IndexingService.start(), let them run 100 milliseconds, then call IndexingService.stop() to send an interruption to producer, causing it to put a poison pill to the queue before exits, causing the consumer to act upon the poison pill and exits.
CHAP7>cat IndexingService.java
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.*;
/**
* IndexingService
* <p/>
* Shutdown with poison pill
*/
public class IndexingService {
private static final int CAPACITY = 1000;
private static final File POISON = new File("");
private final IndexerThread consumer = new IndexerThread();
private final CrawlerThread producer = new CrawlerThread();
private final BlockingQueue<File> queue;
private final FileFilter fileFilter;
private final File root;
public static void main(String args[]) throws InterruptedException {
IndexingService service = new IndexingService(new File("."), f -> f.getName().contains("java"));
service.start();
Thread.sleep(100);
service.stop();
service.awaitTermination();
}
public IndexingService(File root, final FileFilter fileFilter) {
this.root = root;
this.queue = new LinkedBlockingQueue<File>(CAPACITY);
this.fileFilter = new FileFilter() {
public boolean accept(File f) {
return f.isDirectory() || fileFilter.accept(f);
}
};
}
private boolean alreadyIndexed(File f) {
return false;
}
class CrawlerThread extends Thread {
public void run() {
try {
crawl(root);
} catch (InterruptedException e) { /* fall through */
} finally {
while (true) {
try {
queue.put(POISON);
break;
} catch (InterruptedException e1) { /* retry */
}
}
}
}
private void crawl(File root) throws InterruptedException {
File[] entries = root.listFiles(fileFilter);
if (entries != null) {
for (File entry : entries) {
if (entry.isDirectory())
crawl(entry);
else if (!alreadyIndexed(entry))
queue.put(entry);
}
}
}
}
class IndexerThread extends Thread {
public void run() {
try {
while (true) {
File file = queue.take();
if (file == POISON)
break;
else
indexFile(file);
}
} catch (InterruptedException consumed) {
}
}
public void indexFile(File file) {
System.out.println(file.getName());
};
}
public void start() {
producer.start();
consumer.start();
}
public void stop() {
producer.interrupt();
}
public void awaitTermination() throws InterruptedException {
consumer.join();
}
}
CHAP7>javac IndexingService.java
CHAP7>java IndexingService
BetterLogService.java
BrokenPrimeproducer.java
BiConsumerDemo.java
BiFunctionDemo.java
BinaryOperatorDemo.java
BiPredicateDemo.java
ConsumerDemo.java
FunctionDemo.java
PredicateDemo.java
SupplierDemo.java
test.java
UnaryOperatorDemo.java
Compare.java
ConsumerDemo.java
DemoWebServer.java
FutureRenderer.java
IndexingService.java
LifecycleWebServer.java
LogService.java
LogWriter.java
Mix.java
NoncancelableTask.java
OutOfTime.java
OutOfTimeFix.java
Plant.java
PrimeGenerator.java
PrimeProducer.java
ReaderThread.java
Renderer.java
RenderWithTimeBudget.java
SingleThreadRender.java
SingleThreadWebServer.java
SocketUsingTask.java
Basket.java
Birds.java.bk
Box.java
Boxes.java
Cats.java
Fruits.java
Node.java
ObjWild.java
OuterClass.java
Printer.java
SuperConfusing.java
Swimmers.java
TaskExecutionWebServer.java
Test.java
ThreadPerTaskExecutor.java
ThreadPerTaskWebServer.java
TimeBudget.java
TimedRun.java
TimedRun1.java
TimedRun2.java
WithinThreadExecutor.java
CHAP7>
