Форум программистов [Powered by Invision Power Board]

Версия для печати темы
Нажмите сюда для просмотра этой темы в оригинальном формате

Форум программистов > Java: Общие вопросы > Запрет на добавление одинаковых тасков в пул.

Автор: novichiok 21.10.2011, 17:34

Здравствуйте. Стоит такая задача: - Запретить доабавление тасков которые уже имеются в очереди задач для пула. Подробнее:
В приложении создан фиксед-пул, с заданной очередью(ArrayBlockingQueue с фиксированным размером).В пул сабмитятся Runnabl'ы,сабмитятся из разных потоков . В этих Runnabl'ах есть поле id. Мне нужно сделать так, чтобы Runnabl не мог быть добавлен в очередь если в ней уже есть Runnabl с таким же id. При чем уникальная задача должна быть не *все* время работы а только в очереди. То есть допустим , засабмитили Runnable с id1, он выполнился, удалился из очереди, и теперь снова можно сабмитить Runnable с id1.
Как это правильно, канонично реализовать ?
Отнаследоваться от BlockingQueue и переопределить метод который дергает пуловский submit() при добавлении тасков ? В таком случае боюсь что-то поломать в синхронизации самой очереди..

Автор: priam220 21.10.2011, 18:19

не притендую на каноничность, но можно сделать так:

Hashmap<Integer, Runnable> m;

synchronyzed(m){
if(!m.contains(id)){
m.put(id,new Runnable())
}
}

Автор: novichiok 21.10.2011, 19:46

Ну? а в пул как пихать?
PS:у мапа и так функциональность Set'а по ключам

Автор: Stolzen 22.10.2011, 17:57

Ну так сделать свою реализацию пула, которая перед "пиханием" проверяет наличие, и в случае чего реджектит таск.

Добавлено через 10 минут и 8 секунд

Код


public static void main(String[] args) {
    ExecutorService executor = new RejectorExecutor(Executors.newFixedThreadPool(10));
}

static class RejectorExecutor implements ExecutorService {

    private final ExecutorService service;

    public RejectorExecutor(ExecutorService service) {
        this.service = service;
    }

    private final Set<Runnable> set = Collections.synchronizedSet(new HashSet<Runnable>());
    
    public void execute(Runnable command) {
        if (set.contains(command)) {
            throw new RuntimeException("The command was already added");
        }
        
        service.execute(command);
        set.add(command);
    }

И т.д...

Но нужно для реализаций Runnable корректно определить equals и hashCode

Автор: novichiok 24.10.2011, 13:30

Спасибо за ответ. *НО* в таком случае, как Вы описали обеспечивается уникальность тасков всегда, даже после его выполнения. Мне это не подходит, нужно чтоб в каждый момент времени в очереди задач находились только уникальные таски. Если таск был засабмичен и выполнен, то его можно засабмитить снова.

Автор: Stolzen 24.10.2011, 14:49

Так сделайте, чтобы после успешного выполнения таск убирал себя из этого сета.

Автор: Kangaroo 26.10.2011, 00:14

Или (только наследовать надо ArrayBlockingQueue):

Цитата(novichiok @ 21.10.2011, 17:34

)

Отнаследоваться от BlockingQueue и переопределить метод который дергает пуловский submit() при добавлении тасков ?

или использовать композицию, как описано http://codeidol.com/java/java-concurrency/Composing-Objects/Adding-Functionality-to-Existing-Thread-safe-Classes/.

Автор: novichiok 26.10.2011, 17:30

Спасибо, да, Дуга Ли нужно почитать

Автор: LSD 26.10.2011, 17:40

Цитата(Kangaroo @ 26.10.2011, 01:14

)

Или (только наследовать надо ArrayBlockingQueue):

Не выйдет. Судя по описанию задачи, пока задача выполняется добавлять новую нельзя.

Автор: novichiok 26.10.2011, 18:42

Цитата(LSD @ 26.10.2011, 17:40

)

Не выйдет. Судя по описанию задачи, пока задача выполняется добавлять новую нельзя.

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

Автор: Stolzen 27.10.2011, 09:55

Код


public static void main(String[] args) {
    ExecutorService executor = new RejectorExecutor(Executors.newFixedThreadPool(10));
}

static class RejectorExecutor implements ExecutorService {
    private final ExecutorService service;
    private final Set<Runnable> set = Collections.synchronizedSet(new HashSet<Runnable>());
    
    public void execute(Runnable command) {
        if (set.contains(command)) {
            throw new RuntimeException("The command was already added");
        }
        
        service.execute(new RemovableRunnable(command));
        set.add(command);
    }
    
    class RemovableRunnable implements Runnable {
        private final Runnable runnable;

        public RemovableRunnable(Runnable runnable) {
            this.runnable = runnable;
        }
        
        @Override
        public void run() {
            try {
                runnable.run();
            } finally {
                set.remove(runnable);
            }
        }
    }

Автор: LSD 27.10.2011, 10:57

Цитата(novichiok @ 26.10.2011, 19:42

)

Удаляет и выполняет. В противном случае трудно сделать так, чтобы одну таску выполнял только один поток.

У ThreadPoolExecutor есть методы

Код


protected void beforeExecute(Thread t, Runnable r)

protected void afterExecute(Runnable r, Throwable t)

Второй как раз хорошо подойтет для удаления таски из distinct списка.

Автор: novichiok 27.10.2011, 15:04

Всех благодарю за помощь. Действительно к данной проблеме есть множество решений. Я у себя немного переделал логику и исключил возможность добавления дубликатов уже не на уровне пула или очереди