Отмена запроса потребителя производителя

Question

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
}
}

Я бы хотел обработать сценарий, когда пользователь запрашивает завершение работы, нажав Ctrl + C. Я думаю, что это может быть сделано в крюке отключения Однако, как и в приведенном выше коде, каждый производитель получает список идентификаторов (250 может быть?) Для обработки, т. Е. Вызывает БД для получения данных для идентификатора и отправки данных в поток потребителя, который затем обращается к сети. оказание услуг.

Как выйти из цикла for в каждом потоке источника , когда было запрошено завершение работы, чтобы каждый поток не обрабатывал идентификаторы, которые еще не были обработаны? Мне удалось заставить работать shutDownHook, но я не был уверен, как каждый поток должен включать логику в метод run для выхода из метода run () в случае запроса на завершение работы. Может быть, установив внешнюю логическую переменную (AtomicBoolean), чтобы каждый поток проверял цикл for перед обработкой каждого идентификатора?

Насколько я понимаю, если я вызываю shutdown (), он выполняет все представленные задачи и затем завершается. В этом случае невозможно остановить обработку, поскольку задачи уже поставлены в очередь в службе executor.

Если я вызываю shutdownNow () вместо shutdown (), это может привести к неожиданным результатам?

Answer 1

Если вы используете Java SE 6+, у вас есть доступ к JMX классам.Мы посчитали полезным использовать это для «закрытия» запущенных нами служб.

По сути, вы регистрируете свою службу как службу JMX.И используйте класс драйвера для его запуска.

В вашем реальном классе обслуживания реализуйте свою логику, которая по существу использует бесконечный цикл для проверки условия, чтобы увидеть, является ли оно истинным (что будет по умолчанию).

Создайте другой класс драйвера, который подключается к вашей службе JMX и изменяет значение условия цикла на ложное.Затем, когда вы попадете в условие цикла, оно (изящно) отключится и не обработает следующий набор значений.

Answer 2

shutdownNow против shutdown зависит от того, хотите ли вы, чтобы выполняющиеся в данный момент задачи выполнялись первыми.

Если вы хотите, чтобы они немедленно остановились, вы бы сделали две вещи. Сначала вызовите shutdownNow. И во-вторых, в методе run вы проверяете состояние прерывания потока.

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));
        }
    }
}

Здесь вы видите, что метод run теперь знает, что текущий поток был прерван. Затем этот метод запуска может очистить любые данные и завершить работу