public class TaskBlocking {
    /** Монитор. */
    private static final Object mux = new Object();
    
    /** Задачи. */
    private static final Map<Integer, TaskRunnable> tasks = new HashMap<>();

    /**
     * Добавить задачу.
     * 
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;
        
        TaskRunnable taskRun;
        
        synchronized (mux) {
            taskRun = tasks.get(id);
            
            if (taskRun != null)
                taskRun.addTask(task);
            else {
                taskRun = new TaskRunnable(id, task);
                
                isNew = true;
            }
        }
        
        if (isNew)
            submit(taskRun);
    }

    /**
     * Подождать окончания выполнения задач для данного идентификатора.
     * 
     * @param id Идентификатор.
     * @throws InterruptedException В случае прерывания.
     */
    public static void awaitCompletion(Integer id) throws InterruptedException {
        TaskRunnable taskRun;
        
        synchronized (mux) {
            taskRun = tasks.get(id);
            
            if (taskRun == null)
                return; // Нет активных задач.
        }
        
        taskRun.awaitCompletion();
    }

    /**
     * Отдать TaskRunnable в Executor. 
     * 
     * @param taskRun TaskRunnable. 
     */
    private static void submit(TaskRunnable taskRun) {
        /*
            Тут кладете Runnable в Executor. Делаем это вне synchronized, так как в зависимости от конфигурации пула,
            задача может начать выполнение в этом же потоке. 
            Как быть с RejectedExecutionException, решайте сами.
        */
    }
    
    /**
     * Класс, выполняющий задачи.
     */
    private static class TaskRunnable implements Runnable {
        /** Идентификатор. */
        private final int id;

        /** Latch, означающий окончание выполнение задач для текушего id. */
        private final CountDownLatch completeLatch = new CountDownLatch(1);
        
        /** Текущая задача. */
        private Task task;

        /**
         * Конструктор.
         * 
         * @param id Идентификатор.
         * @param task Задача.
         */
        public TaskRunnable(int id, Task task) {
            assert Thread.holdsLock(mux); // Должны быть внутри synchronized для видимости.
            
            this.id = id;
            this.task = task;
        }

        /**
         * Добавить следующую задачу.
         * 
         * @param task Следующая задача.
         */
        public void addTask(Task task) {
            synchronized (mux) {
                this.task = task;
            }
        }
        
        /** {@inheritDoc} */
        @Override 
        public void run() {
            while (true) {
                Task task0;
                
                synchronized (mux) {
                    task0 = task;
                    
                    if (task0 == null) {
                        // Все, больше нет задач в очереди.
                        tasks.remove(id);
                        
                        completeLatch.countDown(); 
                        
                        return;
                    }
                    else
                        task = null; // Занулим следующую задачу.
                }
                
                try {
                    task0.run(); // Выполнить задачу.
                }
                catch (Exception e) {
                    // Обработка на ваше усмотрение.
                }
            }
        }

        /**
         * Подождать окончания выполнения задачи.
         *
         * @throws InterruptedException В случае прерывания.
         */
        public void awaitCompletion() throws InterruptedException {
            completeLatch.await();
        }
    }

    /**
     * Ваша задача.
     */
    private static class Task implements Runnable {
        /** {@inheritDoc} */
        @Override 
        public void run() {
            // Логика задачи.
        }
    }
}

    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;

        TaskRunnable taskRun;

        synchronized (mux) {
            taskRun = tasks.get(id);

            if (taskRun != null)
                taskRun.addTask(task);
            else {
                taskRun = new TaskRunnable(id, task);

                isNew = true;
            }
        }

        if (isNew)
            submit(taskRun);
    }

    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;

        TaskRunnable taskRun;

        synchronized (mux) {
            taskRun = tasks.get(id);

            if (taskRun != null)
                taskRun.addTask(task);
            else {
                taskRun = new TaskRunnable(id, task);

                TaskRunnable oldTaskRun = tasks.put(id, taskRun);
                
                assert oldTaskRun == null;
                
                isNew = true;
            }
        }

        if (isNew)
            submit(taskRun);
    }

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TaskBlocking {
    /** Монитор. */
    private static final Object mux = new Object();
    
    /** Задачи. */
    private static final Map<Integer, TaskRunnable> tasks = new HashMap<Integer, TaskRunnable>();
    
    static ExecutorService e = Executors.newFixedThreadPool(10);

    public static void main(String[] args) {
    	Task t = new Task();
    	Task t2 = new Task();
    	addTask(1, t);
    	addTask(1, t2);
    }
    
    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;

        TaskRunnable taskRun;
        System.out.println("Trying add task");
        synchronized (mux) {
        	System.out.println("Get lock");
            taskRun = tasks.get(id);

            if (taskRun != null) {
            	System.out.println("Task already exist");
                taskRun.addTask(task);
            }
            else {
            	System.out.println("Put new task");
                taskRun = new TaskRunnable(id, task);

                TaskRunnable oldTaskRun = tasks.put(id, taskRun);
                
                assert oldTaskRun == null;
                
                isNew = true;
            }
        }
        System.out.println("Before end");
        if (isNew){
        	submit(taskRun);
        	System.out.println("Task submitted");
        }
        System.out.println("Add finished");
    }

    /**
     * Подождать окончания выполнения задач для данного идентификатора.
     * 
     * @param id Идентификатор.
     * @throws InterruptedException В случае прерывания.
     */
    public static void awaitCompletion(Integer id) throws InterruptedException {
        TaskRunnable taskRun;
        
        synchronized (mux) {
            taskRun = tasks.get(id);
            
            if (taskRun == null)
                return; // Нет активных задач.
        }
        
        taskRun.awaitCompletion();
    }

    /**
     * Отдать TaskRunnable в Executor. 
     * 
     * @param taskRun TaskRunnable. 
     */
    private static void submit(TaskRunnable taskRun) {
    	e.submit(taskRun);
        /*
            Тут кладете Runnable в Executor. Делаем это вне synchronized, так как в зависимости от конфигурации пула,
            задача может начать выполнение в этом же потоке. 
            Как быть с RejectedExecutionException, решайте сами.
        */
    }
    
    /**
     * Класс, выполняющий задачи.
     */
    private static class TaskRunnable implements Runnable {
        /** Идентификатор. */
        private final int id;

        /** Latch, означающий окончание выполнение задач для текушего id. */
        private final CountDownLatch completeLatch = new CountDownLatch(1);
        
        /** Текущая задача. */
        private Task task;

        /**
         * Конструктор.
         * 
         * @param id Идентификатор.
         * @param task Задача.
         */
        public TaskRunnable(int id, Task task) {
            assert Thread.holdsLock(mux); // Должны быть внутри synchronized для видимости.
            
            this.id = id;
            this.task = task;
        }

        /**
         * Добавить следующую задачу.
         * 
         * @param task Следующая задача.
         */
        public void addTask(Task task) {
            synchronized (mux) {
                this.task = task;
            }
        }
        
        /** {@inheritDoc} */
        @Override 
        public void run() {
            while (true) {
                Task task0;
                
                synchronized (mux) {
                    task0 = task;
                    
                    if (task0 == null) {
                        // Все, больше нет задач в очереди.
                        tasks.remove(id);
                        
                        completeLatch.countDown(); 
                        
                        return;
                    }
                    else
                        task = null; // Занулим следующую задачу.
                }
                
                try {
                    task0.run(); // Выполнить задачу.
                }
                catch (Exception e) {
                    // Обработка на ваше усмотрение.
                }
            }
        }

        /**
         * Подождать окончания выполнения задачи.
         *
         * @throws InterruptedException В случае прерывания.
         */
        public void awaitCompletion() throws InterruptedException {
            completeLatch.await();
        }
        

    }

    /**
     * Ваша задача.
     */
    private static class Task implements Runnable {
        /** {@inheritDoc} */
        @Override 
        public void run() {
            System.out.println("Task started");
            try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
            System.out.println("Task stopped");
        }
    }
}

public class TaskBlocking {
    /** Монитор. */
    private static final Object mux = new Object();

    /** Задачи. */
    private static final Map<Integer, TaskRunnable> tasks = new HashMap<Integer, TaskRunnable>();

    static ExecutorService e = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws Exception {
        addTask(1, new Task(1));

        // Thread.sleep(200); // Расскомментируйте, и увидите, что выполнится и первый, и второй.

        addTask(1, new Task(2));
    }

    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;

        TaskRunnable taskRun;
        System.out.println("Trying add task");
        synchronized (mux) {
            System.out.println("Get lock");
            taskRun = tasks.get(id);

            if (taskRun != null) {
                System.out.println("Task already exist");
                taskRun.addTask(task);
            }
            else {
                System.out.println("Put new task");
                taskRun = new TaskRunnable(id, task);

                TaskRunnable oldTaskRun = tasks.put(id, taskRun);

                assert oldTaskRun == null;

                isNew = true;
            }
        }
        System.out.println("Before end");
        if (isNew){
            submit(taskRun);
            System.out.println("Task submitted");
        }
        System.out.println("Add finished");
    }

    /**
     * Подождать окончания выполнения задач для данного идентификатора.
     *
     * @param id Идентификатор.
     * @throws InterruptedException В случае прерывания.
     */
    public static void awaitCompletion(Integer id) throws InterruptedException {
        TaskRunnable taskRun;

        synchronized (mux) {
            taskRun = tasks.get(id);

            if (taskRun == null)
                return; // Нет активных задач.
        }

        taskRun.awaitCompletion();
    }

    /**
     * Отдать TaskRunnable в Executor.
     *
     * @param taskRun TaskRunnable.
     */
    private static void submit(TaskRunnable taskRun) {
        e.submit(taskRun);
        /*
            Тут кладете Runnable в Executor. Делаем это вне synchronized, так как в зависимости от конфигурации пула,
            задача может начать выполнение в этом же потоке.
            Как быть с RejectedExecutionException, решайте сами.
        */
    }

    /**
     * Класс, выполняющий задачи.
     */
    private static class TaskRunnable implements Runnable {
        /** Идентификатор. */
        private final int id;

        /** Latch, означающий окончание выполнение задач для текушего id. */
        private final CountDownLatch completeLatch = new CountDownLatch(1);

        /** Текущая задача. */
        private Task task;

        /**
         * Конструктор.
         *
         * @param id Идентификатор.
         * @param task Задача.
         */
        public TaskRunnable(int id, Task task) {
            assert Thread.holdsLock(mux); // Должны быть внутри synchronized для видимости.

            this.id = id;
            this.task = task;
        }

        /**
         * Добавить следующую задачу.
         *
         * @param task Следующая задача.
         */
        public void addTask(Task task) {
            synchronized (mux) {
                this.task = task;
            }
        }

        /** {@inheritDoc} */
        @Override
        public void run() {
            while (true) {
                Task task0;

                synchronized (mux) {
                    task0 = task;

                    if (task0 == null) {
                        // Все, больше нет задач в очереди.
                        tasks.remove(id);

                        completeLatch.countDown();

                        return;
                    }
                    else
                        task = null; // Занулим следующую задачу.
                }

                try {
                    task0.run(); // Выполнить задачу.
                }
                catch (Exception e) {
                    // Обработка на ваше усмотрение.
                }
            }
        }

        /**
         * Подождать окончания выполнения задачи.
         *
         * @throws InterruptedException В случае прерывания.
         */
        public void awaitCompletion() throws InterruptedException {
            completeLatch.await();
        }


    }

    /**
     * Ваша задача.
     */
    private static class Task implements Runnable {

        public final int id;

        private Task(int id) {
            this.id = id;
        }

        /** {@inheritDoc} */
        @Override
        public void run() {
            System.out.println("Task started [id=" + id + ']');
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            System.out.println("Task stopped [id=" + id + ']');
        }
    }
}

public class TaskBlockingAndQueued {
    /** Монитор. */
    private static final Object mux = new Object();

    /** Задачи. */
    private static final Map<Integer, TaskRunnable> tasks = new HashMap<Integer, TaskRunnable>();

    /** Очередь задач. */
    private static final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();

    /** Пул. */
    private static final ExecutorService e = new ThreadPoolExecutor(8, 8, 1000L, TimeUnit.MILLISECONDS, taskQueue);

    public static void main(String[] args) throws Exception {
        addTask(1, new Task(1));

        //Thread.sleep(200); // Расскомментируйте, и увидите, что выполнится и первый, и второй.

        addTask(1, new Task(2));

        e.shutdown();
    }

    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(Integer id, Task task) {
        boolean isNew = false;

        TaskRunnable taskRun;

        synchronized (mux) {
            taskRun = tasks.get(id);

            if (taskRun != null)
                taskRun.addTask(task);
            else {
                System.out.println("Put new task");
                taskRun = new TaskRunnable(id, task);

                TaskRunnable oldTaskRun = tasks.put(id, taskRun);

                assert oldTaskRun == null;

                isNew = true;
            }
        }

        if (isNew)
            submit(taskRun);
    }

    /**
     * Подождать окончания выполнения задач для данного идентификатора.
     *
     * @param id Идентификатор.
     * @throws InterruptedException В случае прерывания.
     */
    public static void awaitCompletion(Integer id) throws InterruptedException {
        TaskRunnable taskRun;

        synchronized (mux) {
            taskRun = tasks.get(id);

            if (taskRun == null)
                return; // Нет активных задач.
        }

        taskRun.awaitCompletion();
    }

    /**
     * Отдать TaskRunnable в Executor.
     *
     * @param taskRun TaskRunnable.
     */
    private static void submit(TaskRunnable taskRun) {
        try {
            taskQueue.put(taskRun);
        }
        catch (InterruptedException e) {
            // Handle.
        }
    }

    /**
     * Класс, выполняющий задачи.
     */
    private static class TaskRunnable implements Runnable {
        /** Идентификатор. */
        private final int id;

        /** Latch, означающий окончание выполнение задач для текушего id. */
        private final CountDownLatch completeLatch = new CountDownLatch(1);

        /** Текущая задача. */
        private Task task;

        /**
         * Конструктор.
         *
         * @param id Идентификатор.
         * @param task Задача.
         */
        public TaskRunnable(int id, Task task) {
            assert Thread.holdsLock(mux); // Должны быть внутри synchronized для видимости.

            this.id = id;
            this.task = task;
        }

        /**
         * Добавить следующую задачу.
         *
         * @param task Следующая задача.
         */
        public void addTask(Task task) {
            synchronized (mux) {
                this.task = task;
            }
        }

        /** {@inheritDoc} */
        @Override
        public void run() {
            while (true) {
                boolean proceed = false;

                Task task0;

                synchronized (mux) {
                    task0 = task;

                    if (task0 == null) {
                        // Все, больше нет задач в очереди.
                        tasks.remove(id);

                        completeLatch.countDown();

                        return;
                    }
                    else {
                        if (canProceed()) {
                            proceed = true;

                            task = null; // Занулим следующую задачу.
                        }
                    }
                }

                if (proceed) {
                    try {
                        task0.run(); // Выполнить задачу.
                    }
                    catch (Exception e) {
                        // Обработка на ваше усмотрение.
                    }
                }
                else {
                    returnToQueue();
                    
                    return;
                }
            }
        }

        /**
         * Подождать окончания выполнения задачи.
         *
         * @throws InterruptedException В случае прерывания.
         */
        public void awaitCompletion() throws InterruptedException {
            completeLatch.await();
        }

        /**
         * Проверить, можно ли продолжать выполнение задач, или надо вернуть объект в конец очереди.
         *
         * @return {@code true}, если
         */
        private boolean canProceed() {
            // Ваша имплементация.
            return true;
        }

        /**
         * Вернуть задачу в очередь.
         */
        private void returnToQueue() {
            // Ваша исплементация, напр.:
            try {
                taskQueue.put(this);
            }
            catch (InterruptedException e) {
                // Handle.
            }
        }
        
    }

    /**
     * Ваша задача.
     */
    private static class Task implements Runnable {

        private final int id;

        private Task(int id) {
            this.id = id;
        }

        /** {@inheritDoc} */
        @Override
        public void run() {
            System.out.println("Task started [id=" + id + "]");

            try {
                Thread.sleep(2000);
            }
            catch (InterruptedException e) {
                // Handle.
            }

            System.out.println("Task finished [id=" + id + "]");
        }
    }
}

public class TestBlocking {
    /** Монитор. */
    private static final Object mux = new Object();

    /** Задачи. */
    private static final Map<Integer, Task> taskMap = new HashMap<>();

    /**
     * Добавить задачу.
     *
     * @param id Идентификатор.
     * @param task Задача.
     */
    public static void addTask(int id, Task task) {
        synchronized (mux) {
            Task previousTask = taskMap.get(id);

            if (previousTask != null) {
                boolean cancelled = previousTask.tryCancel(); // Пробуем отменить последнюю задачу.

                if (cancelled)
                    // Получилось, сохраним референс на текущую активную задачу.
                    task.setPreviousTask(previousTask.getPreviousTask());
                else
                    // Не получилось, значит последняя задача является активной.
                    task.setPreviousTask(previousTask);
            }

            taskMap.put(id, task);
        }

        try {
            // Каждая новая задача идет в отдельном Runnable.
            submit(task);
        }
        catch (RejectedExecutionException e) {
            // Обработать.
        }
    }

    /**
     * Подождать завершения последней поступившей задачи.
     *
     * @param id Идентификатор.
     * @throws InterruptedException Если произошло прерывание.
     */
    public static void awaitCompletion(int id) throws InterruptedException {
        Task task;

        synchronized (mux) {
            task = taskMap.get(id);
        }

        // Вновь поступающие задачи не влияют на завершение данной задачи. Starvation отсутствует.
        if (task != null)
            task.awaitCompletion();
    }

    /**
     * Отдать задачу в Executor.
     *
     * @param task Задача.
     */
    private static void submit(Task task) {
        // ...
    }

    /**
     * Задача.
     */
    private class Task implements Runnable {
        /** Идентификатор. */
        private final int id;

        /** Latch, сигнализирующий окончание работы задачи. */
        private final CountDownLatch completeLatch = new CountDownLatch(1);

        /** Предыдущая задача, которая была активна в момент добавления данной задачи. */
        private Task previousTask;

        /** Флаг начала выполнения задачи. */
        private boolean started;

        /** Флаг отмены задачи. */
        private boolean cancelled;

        /**
         * @param id Идентификатор.
         */
        private Task(int id) {
            this.id = id;
        }

        /**
         * Получить предыдущую задачу.
         *
         * @return Предыдущая задача.
         */
        private Task getPreviousTask() {
            synchronized (mux) {
                return previousTask;
            }
        }

        /**
         * Установить предыдущую задачу.
         *
         * @param previousTask Предыдущая задача.
         */
        private void setPreviousTask(Task previousTask) {
            synchronized (mux) {
                assert this.previousTask == null;

                this.previousTask = previousTask;
            }
        }

        /**
         * Подождать окончания работы задачи.
         *
         * @throws InterruptedException В случае прерывания.
         */
        public void awaitCompletion() throws InterruptedException {
            Task previousTask0;

            synchronized (mux) {
                previousTask0 = previousTask;
            }

            if (previousTask0 != null)
                previousTask0.awaitCompletion();

            completeLatch.await();
        }

        /**
         * Попробовать отменить задачу.
         *
         * @return {@code true}, если задача была отменена; {@code false}, если задачу нельзя отменить, так как она
         *     уже начала выполняться.
         */
        public boolean tryCancel() {
            synchronized (mux) {
                if (started)
                    return false; // Уже нельзя отменить.
                else {
                    cancelled = true;

                    completeLatch.countDown();

                    return true; // Успели отменить.
                }
            }
        }

        /** {@inheritDoc} */
        @Override
        public void run() {
            Task previousTask;

            synchronized (mux) {
                if (cancelled)
                    return;

                started = true;

                previousTask = this.previousTask;

                this.previousTask = null; // Что бы избежать memory leak.
            }

            try {
                previousTask.awaitCompletion();

                run0();
            }
            catch (InterruptedException e) {
                // Обработать.
            }
            finally {
                synchronized (mux) {
                    Task task0 = taskMap.get(id);

                    if (task0 == this) // Что бы не удалить более новую задачу.
                        taskMap.remove(id);

                    completeLatch.countDown();
                }
            }
        }

        /**
         * Логика задачи.
         */
        private void run0() {
            // ...
        }
    }
}