CODEWITHSUNDEEP
javamultithreadingconcurrencyproducer-consumerexecutorservice
Oxford
Oxford

Reputation: 139

Producer consumer request cancellation

public class MainClass {

private static final int size = 5;

private ExecutorService prodExec = Executors.newFixedThreadPool(size);
private ExecutorService consExec = Executors.newFixedThreadPool(size);

//main method here

public void start(String[] args) {

    for (int index = 0; index < size; index++) {
        Runnable producer = new Producer(consExec, listOfIds);
        prodExec.execute(producer);
    }

    Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
            prodExec.shutdown();
            try {
                prodExec.awaitTermination(10, TimeUnit.SECONDS);
            } catch (InterruptedException ignore) {
            }

            consExec.shutdown();
            try {
                consExec.awaitTermination(10, TimeUnit.SECONDS);
            } catch (InterruptedException ignore) {
            }
        }
    });
    }
}
public class Producer implements Runnable {

private ExecutorService consExec;
private List<Long> list;

public Producer(ExecutorService exec, List<Long> list) {
    this.consExec = exec;
    this.list = list;
}

public void run() {
    for (Long id: list) {
        data = get data from db for the id
        consExec.execute(new Consumer(data));
    }
}
}
public class Consumer implements Runnable {

public void run() {
    // call web service
}
}

I would like to handle the scenario when a user requests a shutdown may be by pressing Ctrl+C. I think this could be done in the shutdown hook. However, as in the above code, each Producer gets a list of IDs (250 may be?) to process i.e., call the db to retrieve data for the ID and submit the data to the consumer thread, which then calls out to a web service.

How do I break out of the for loop in each of the Producer thread, when a shutdown has been requested so that each thread does not process the ids that haven't been processes yet? I was able to get the shutDownHook to work, but not sure how each thread should incorporate the logic within the run method to exit the run() method in case of a shutdown request. May be by setting a boolean variable(AtomicBoolean) externally that each thread checks in the for loop before processing every id?

As I understand it, if I call shutdown(), it executes all the submitted tasks and then finishes. In this case, there is no way of stopping processing as the tasks have already beeb queued on the executor service.

If I call shutdownNow() instead of shutdown() it might produce unexpected results?

Upvotes: 1

Views: 521

Answers (2)

Chris Aldrich
Chris Aldrich

Reputation: 1932

If you are using Java SE 6+ you have access to the JMX classes. We found it helpful to use this to "shut down" services that we have running.

In essence you register your service as a JMX service. And use a driver class to fire it off.

In your actual service class, implement your logic that in essence uses an infinite loop to check a condition to see if it is true (which it will be by default).

Create another driver class that connects to your JMX service and modifies the value of the loop condition to be false. Then when you hit your loop condition it will (gracefully) shut down and not process the next set of values.

Upvotes: 1

John Vint
John Vint

Reputation: 40256

shutdownNow vs shutdown depends on if you want the currently running tasks to complete first.

If you want them to stop immidiately you would do two things. First invoke shutdownNow. And two, within the run method you test the thread's interruption status.

public class Producer implements Runnable {

private ExecutorService consExec;
private List<Long> list;

public Producer(ExecutorService exec, List<Long> list) {
    this.consExec = exec;
    this.list = list;
}

    public void run() {
        for (Long id: list) {
            if(Thread.currentThread().isInterrupted()){
               //the shutdownNow method has been called (or may a future.cancel(true))
            }
            data = get data from db for the id
            consExec.execute(new Consumer(data));
        }
    }
}

You can see here the the run method now knows the current thread has been interrupted. That run method then can clean up any data and exit

Upvotes: 1

Related Questions