java: executors + tasks + locks

Question

Предположим, что у меня есть ExecutorService (который может быть пулом потоков, поэтому есть параллелизм), который выполняет задачу в разное время, периодически или в ответ на какое-то другое условие. Задача, которая должна быть выполнена, следующая:

• Если эта задача уже выполняется, ничего не предпринимайте (и пусть закончившаяся работа заканчивается).
• Если эта задача еще не выполняется, запустите Алгоритм X, который может занять много времени.

Я пытаюсь придумать способ реализации этого. Это должно быть что-то вроде:

Runnable task = new Runnable() { 
    final SomeObj inProgress = new SomeObj(); 
    @Override public void run() { 
     if (inProgress.acquire()) 
     { 
      try 
      { 
      algorithmX(); 
      } 
      finally 
      { 
      inProgress.release(); 
      } 
     } 
    } 
} 

// re-use this task object whenever scheduling the task with the executor 

где SomeObj является либо ReentrantLock (приобретение = tryLock() и отпустить = unlock()) или AtomicBoolean или что-то, но я не уверен,. Мне нужен ReentrantLock здесь? (Возможно, я хочу блокировку без реентера в случае, если algorithmX() заставляет эту задачу запускать рекурсивно!) Или достаточно было бы AtomicBoolean?

---

изменить: для блокировки, не являющейся реентерабельным, это подходит?

Runnable task = new Runnable() { 
    boolean inProgress = false; 
    final private Object lock = new Object(); 
    /** try to acquire lock: set inProgress to true, 
    * return whether it was previously false 
    */ 
    private boolean acquire() { 
     synchronized(this.lock) 
     { 
     boolean result = !this.inProgress; 
     this.inProgress = true; 
     return result; 
     } 
    } 
    /** release lock */ 
    private void release() { 
     synchronized(this.lock) 
     { 
     this.inProgress = false; 
     } 
    } 
    @Override public void run() { 
     if (acquire()) 
     { 
      // nobody else is running! let's do algorithmX() 
      try 
      { 
      algorithmX(); 
      } 
      finally 
      { 
      release(); 
      } 
     } 
     /* otherwise, we are already in the process of 
     * running algorithmX(), in this thread or in another, 
     * so don't do anything, just return control to the caller. 
     */ 
    } 
} 

Answer 1

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

Ниже является гораздо более эффективной реализации с теми же неподходящих слабостей использовать в качестве реализации:

AtomicBoolean inProgress = new AtomicBoolean(false) 
    /* Returns true if we acquired the lock */ 
    private boolean acquire() { 
     return inProgress.compareAndSet(false, true); 
    } 
    /** Always release lock without determining if we in fact hold it */ 
    private void release() { 
     inProgress.set(false); 
    } 

Answer 2

ReentrantLock кажется прекрасным для меня. Единственная ситуация, когда мне было бы интересно вручную создать блокировку с использованием AtomicInteger, будет, если у вас действительно короткий algorithmX, который не ваш случай.

Answer 3

Ваш первый бит кода выглядит довольно хорошо, но если вы беспокоитесь о algorithmX рекурсивно Призывая задача, я бы предложил использовать файл java.util.concurrent.Semaphore как объект синхронизации, а не ReentrantLock. Например:

Runnable task = new Runnable() { 
    final Semaphore lock = new Semaphore(1); 
    @Override public void run() { 
     if (lock.tryAcquire()) 
     { 
      try 
      { 
      algorithmX(); 
      } 
      finally 
      { 
      lock.release(); 
      } 
     } 
    } 
} 

Примечание, в частности, использование попробоватьприобретают. Если получение блокировки не выполняется, algorithmX не запускается.

Answer 4

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

Что означает «ничего не делать» в этом контексте? Поток должен блокировать и повторять выполнение после запуска algorithmX закончен ?. Если это так, то semaphore.acquire вместо tryAcquire следует использовать, и решение AtomicBoolean не будет работать должным образом.