Флаг остановки в многопоточном обработчике

Question

[Denis Kostousov]

День добрый.
Не смотря на то, что давольно давно пишу на Java, специалистом в области многопоточного программирования не являюсь. Поэтому хочу проконсультироваться у более опытных коллег.
Задача:
Есть некий класс, в многопоточном режиме обрабатывающий запросы (бизнес метод service). Нужно сделать метод остановки (назовем его destroy). Работать должно следующим образом:
Нужно сделать так, что бы после вызова метода destroy новые запросы не обрабатывались, обработка старых благополучно завершилась и после завершения последней обработки выполнились какие-то действия.
Я сделал так:

public class Worker{
    AtomicBoolean flag;
    AtomicInteger counter;

    public void destroy() {
	flag.set(true);
	if (counter.get() == 0) {
	    destroyImpl();
	}
    }

    public void service(String arg) {
	if (!flag.get()) {
	    counter.incrementAndGet();
	    if (!flag.get()) {
		serviceImpl(arg);
	    }
	    counter.decrementAndGet();
	} else {
	    if (counter.get() == 0) {
		destroyImpl();
	    }
	}
    }

}

Подозреваю что не только коряво, но и глючно.
Как правильно?

Comments

[aggrrrh]

Не совсем в тему, но у яндекса есть курс видео лекций по параллельным и распределенным вычислениям. В том числе на java.

Answer 1

[Руслан Лопатин]

Совершенно неправильное использование атомарных операций. Метод destroyImpl(), например, может быть вызван многократно. Почему бы вам не воспользоваться классическими блокировками и не избавиться от атомарных сущностей? Так проще и намного читабельней. К тому же вам необходимо работать сразу с несколькими полями — а это нетривиально, если использовать лишь атомарные операции.

Должно получиться что-то вроде этого:

public class Worker {

    private boolean destroy;
    private boolean destroyed;
    private counter;

    public void destroy() {
        synchronized (this) {
            if (this.destroyed) {
                return;
            }
            this.destroy = true;
            if (this.counter != 0) {
                return;
            }
            this.destroyed = true;
        }
        destroyImpl();
    }

    public void service(String arg) {
        synchronized (this) {
            if (this.destroy) {
                return;
            }
            ++this.counter;
        }
        serviceImpl(arg);
        synchronized (this) {
            if (--this.counter != 0) {
                return;
            }
            if (!this.destroy) {
                return;
            }
            this.destroyed = true;
        }
        destroyImpl()
    }

}

Comments

[Руслан Лопатин]

Ну и, разумеется, вызов serviceImpl(arg) лучше заключить в блок try {..}, а оставшуюся часть метода — в блок finally {...} на случай, если serviceImpl() вызовет исключение.

[Руслан Лопатин]

Блин, забыл проверить флаг destroyed в методе service() перед вызовом destroyImpl().

[FanKiLL]

К переменным желательно приписать volatile тогда, чтоб не кешировались.

[Руслан Лопатин]

Блок synchronized в Java включает также барьеры памяти. Так что volatile в данном случае не нужен.

Answer 2

[FanKiLL]

Используйие ThreadPoolExecutor У него есть метод shutdown() который после вызова закончит «таски», которые вы ему дали на выполнение, но новые принимать больше не будет.

Таски могут быть к примеру или Callable<T> если хотите возвращать значение или Runnable

Comments

[Denis Kostousov]

Многопоточный вызов организую не самостоятельно. Просто подсовываю в сервлет, там «все само». Городить еще один набор потоков как-то не хочется.
Нужен способ правильно остановку своего класса организовать.

Answer 3

[Divers]

Несколько раз прочитал вопрос, но так и не понял — у вас 1 инстанс Worker на много потоков? Если на нем синхронизироваться, то вы же получите обычное последовательное исполнение, зачем тогда вообще многопоточность. Поясните пожалуйста.

Comments

[Руслан Лопатин]

Ну не обязательно же синхронизироваться на всё время исполнения. Синхронизированные блоки кода должны быть как можно короче. Вся остальная логика работает в отдельных потоках.

[Divers]

Все еще не понял — 1 инстанс Worker на все потоки?

[Denis Kostousov]

Да. Один экземпляр. Много потоков.

[Divers]

А, я кажется понял. У вас задача не распараллелить работу, а просто сохранить консистентность данных для каждого потока. Правда если входные данные в сервис не зависят друг от друга, зачем это все в 1 инстанс пихать?

[Denis Kostousov]

Потоками я не управляю. Просто подсунул экземпляр. Сервлеты сами разбираются с потоками и экземплярами. Создавать на каждый запрос обработчик — накладно.

[Divers]

Ну для этого и используют пулы объектов. А один экземпляр на много потоков — это же какая-то бессмыслица. Мне кажется многие ( к примеру FanKiLL сказал про пул, хотя вы вроде как используете 1 объект), в том числе и я до сих пор не понимают что конкретно вы имеете и что хотите получить