I am trying to implement the producer-consumer pattern and I want to be able to stop the consumer. So far I wrote:
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Test {
public static void main(final String[] args) throws InterruptedException {
final BlockingQueue<String> messages = new LinkedBlockingQueue<>(100);
final ExecutorService producer = Executors.newSingleThreadExecutor();
final Consumer consumer = new Consumer(messages);
producer.execute(new Runnable() {
@Override
public void run() {
for (int i = 0;; i++) {
try {
messages.put("Message " + i);
Thread.sleep(500);
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
});
consumer.start();
consumer.stop();
consumer.start();
}
private static class Consumer {
private final ExecutorService consumer = Executors.newSingleThreadExecutor();
private final BlockingQueue<String> messages;
public Consumer(final BlockingQueue<String> m) {
messages = m;
}
public void start() {
consumer.execute(new Runnable() {
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
try {
System.out.println(messages.take());
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}
});
}
public void stop() {
consumer.shutdownNow();
}
}
}
But this always gives me the exception above. The console output is:
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Test$Consumer$1@31602bbc rejected from java.util.concurrent.ThreadPoolExecutor@20d75cf7[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:628)
at Test$Consumer.start(Test.java:46)
at Test.main(Test.java:31)
Why am I not able to "restart" the executor or how can I achieve this?
Thanks in advance!