0

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!

2 Answers 2

1

Documentation says

Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted

So Restart is not possible untill you modify JDK code.

1

This is because when your class Consumer tries to consumer.execute(new Runnable() {, its executor is already shutdowned. Why? Because after line consumer.start(); your program terminates instantly: you don't wait until tasks will be executed.

You should modify your code smthng like that:

  1. use submit method, instead of execute: producer.submit(new Runnable() {...}
  2. receive from execute() method Future<?> object: Future f = producer.submit(new Runnable() { ... }

  3. wait until that object will terminate:

consumer.start();

f.get();

UPDATE According to updated code in question, we should discuss theese two new lines:

    consumer.start();

    consumer.stop();
    consumer.start();

You should known, that after your executor was stopped once, you will be not able to execute new task in it. So this explains why you receive this exception in last line consumer.start();.

1
  • Sorry, I forgot two lines after this line. I have made an update to the question. I call stop and start after the first start call.
    – stevecross
    Commented May 21, 2014 at 14:11

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.