public class TaskHolder {
    /** Максимальное количество запросов в очереди, что бы избежать OOME. */
    private final int maxSize;

    /** Маппинг с идентификатора игрока на группу задач. */
    private final Map<Long, TaskGroup> taskGrps = new LinkedHashMap<>();

    /** Очередь с группами задач. */
    private final BlockingQueue<Runnable> taskGrpQueue = new LinkedBlockingDeque<>();

    /** Общее количество задач в очереди. */
    private int size;

    /**
     * Конструктор.
     *
     * @param maxSize Максимальное количество задач в очереди.
     */
    public TaskHolder(int maxSize) {
        this.maxSize = maxSize;
    }

    /**
     * Добавить задачу в очередь.
     *
     * @param playerId Идентификатор игрока.
     * @param task Задача.
     */
    public synchronized void addTask(Long playerId, Runnable task) throws InterruptedException {
        // Ждем, пока не появится место в очереди.
        while (size >= maxSize)
            wait();

        TaskGroup taskGrp = taskGrps.get(playerId);

        if (taskGrp == null) {
            // Группы, ассоциированной с таким игроком, еще нет.
            taskGrp = new TaskGroup(playerId);

            taskGrp.tasks.add(task);

            taskGrps.put(playerId, taskGrp);

            taskGrpQueue.add(taskGrp);
        }
        else
            // Группа, ассоциированная с таким игроком, есть.
            taskGrp.tasks.add(task);

        size++;
    }

    /**
     * Получить очередь сгруппированных задач.
     *
     * @return Очередь сгруппированных задач.
     */
    public BlockingQueue<Runnable> taskGroupQueue() {
        return taskGrpQueue;
    }

    /**
     * Группа задач для одного игрока.
     */
    private class TaskGroup implements Runnable {
        /** Идентификатор игрока. */
        private final Long playerId;

        /** Задачи в данной группе. */
        private final Deque<Runnable> tasks = new ConcurrentLinkedDeque<>();

        /**
         * Конструктор.
         *
         * @param playerId Идентификатор игрока.
         */
        private TaskGroup(Long playerId) {
            this.playerId = playerId;
        }

        /** {@inheritDoc} */
        @Override public void run() {
            int tasksSize = tasks.size();

            // Вычитываем столько тасков, сколько их было в момент вызова
            // Метода run(). Иначе мы рисуем столкнуться со starvation, когда
            // постоянно прибывающие новые таски от одного игрока подавляют ранее
            // пришедшие таски от другого игрока.
            for (int i = 0; i < tasksSize; i++) {
                Runnable task = tasks.poll();

                synchronized (Handler.this) {
                    size--;

                    notifyAll();
                }

                try {
                    task.run();
                }
                catch (Exception e) {
                    // Обработка эксепшна, пришедшего из таска.
                }
            }

            synchronized (Handler.this) {
                // Если новых задач для данной группы не поступало, удалить ее.
                // Иначе - поставить в конец очереди.
                if (tasks.isEmpty())
                    taskGrps.remove(playerId);
                else
                    taskGrpQueue.add(this);
            }
        }
    }
}