Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Здравствуйте, GreenTea, Вы писали:

GT>>Но это частный случай в котором случайно совпали тайминги добавления и обработки тасок. В общем же случае вероятность такой ситуации очень мала.

EP>Даже если и так — условие всё равно не выполняется. Я бы ещё понял если бы это условие было трудно достичь — так ведь нет, есть же простой вариант циклического перебора клиентов.

Тогда вот так.. Тут трекается история clientId тасок которые выполнялись, в последовательности выполнения, и из свободных берется самая ранняя таская клиента, который обрабатывался раньше всех по истории.
Величина истории ограничена 100, если надо можно вынести это в параметр пула.

package net.sf.brunneng;


import java.util.*;

public class ThreadPool {

   private class ClientTask {
      final int clientId;
      final Runnable runnable;

      public ClientTask(int clientId, Runnable runnable) {
         this.clientId = clientId;
         this.runnable = runnable;
      }
   }

   private static final int MAX_EXECUTED_CLIENTS_HISTORY = 100;

   private boolean closed;
   private final List<ClientTask> tasks = Collections.synchronizedList(new ArrayList<ClientTask>());

   private List<Thread> threads = new ArrayList<Thread>();
   private Set<Integer> executingClients = Collections.synchronizedSet(new HashSet<Integer>());

   private LinkedList<Integer> executedClientsHistory = new LinkedList<Integer>();

   public List<ClientTask> getTasks() {
      return tasks;
   }

   public ThreadPool(int threadsCount) {
      for (int i = 0; i < threadsCount; ++i) {
         Thread thread = createThread();
         threads.add(thread);
         thread.start();
      }
   }

   private ClientTask getNextClientTask() {
      ClientTask res = null;

      synchronized (tasks) {

         Map<Integer, Integer> clientIdToEarliestTaskInstanceMap = new HashMap<Integer, Integer>();
         for (int i = 0; i < tasks.size(); i++) {
            ClientTask task = tasks.get(i);
            if (!executingClients.contains(task.clientId) &&
                    !clientIdToEarliestTaskInstanceMap.containsKey(task.clientId)) {
               clientIdToEarliestTaskInstanceMap.put(task.clientId, i);
            }
         }

         final Map<Integer, Integer> clientIdMinStepsBack = new HashMap<Integer, Integer>();
         Iterator<Integer> iterator = executedClientsHistory.descendingIterator();
         int i = 1;
         while (iterator.hasNext()) {
            Integer clientId = iterator.next();
            if (!clientIdMinStepsBack.containsKey(clientId)) {
               clientIdMinStepsBack.put(clientId, i);
            }
            i++;
         }

         int maxStepBack = -1;
         Integer bestEarliestTaskInstance = null;
         for (Integer clientId : clientIdToEarliestTaskInstanceMap.keySet()) {
            Integer stepsBack = clientIdMinStepsBack.get(clientId);
            if (stepsBack == null) { // no occurence in history
               bestEarliestTaskInstance = clientIdToEarliestTaskInstanceMap.get(clientId);
               break;
            }
            else if (stepsBack > maxStepBack) {
               maxStepBack = stepsBack;
               bestEarliestTaskInstance = clientIdToEarliestTaskInstanceMap.get(clientId);
            }
         }

         if (bestEarliestTaskInstance != null) {
            res = tasks.remove(bestEarliestTaskInstance.intValue());
            executingClients.add(res.clientId);

            executedClientsHistory.add(res.clientId);
            if (executedClientsHistory.size() > MAX_EXECUTED_CLIENTS_HISTORY) {
               executedClientsHistory.removeFirst();
            }
         }
      }

      return res;
   }

   private Thread createThread() {
      return new Thread(new Runnable() {
         @Override
         public void run() {
            boolean finish = false;

            while (!finish && !closed) {
               try {
                  ClientTask task = getNextClientTask();

                  if (task == null) {
                     synchronized (tasks) {
                        tasks.wait();
                     }
                     continue;
                  }

                  task.runnable.run();
                  executingClients.remove(task.clientId);

               } catch (InterruptedException e) {
                  e.printStackTrace();
                  finish = true;
               }
            }
         }
      });
   }

   public void addTask(int clientId, Runnable task) {
      synchronized (tasks) {
         tasks.add(new ClientTask(clientId, task));

         if (!executingClients.contains(clientId)) {
            tasks.notify();
         }
      }
   }

   public void close() {
      closed = true;

      synchronized (tasks) {
         tasks.notifyAll();
      }

      for (Thread thread : threads) {
         try {
            thread.join();
         } catch (InterruptedException ignored) {

         }
      }
   }
}

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Здравствуйте, GreenTea, Вы писали:

GT>>Но это частный случай в котором случайно совпали тайминги добавления и обработки тасок. В общем же случае вероятность такой ситуации очень мала.

EP>Даже если и так — условие всё равно не выполняется. Я бы ещё понял если бы это условие было трудно достичь — так ведь нет, есть же простой вариант циклического перебора клиентов.

Тогда вот так.. Тут трекается история clientId тасок которые выполнялись, в последовательности выполнения, и из свободных берется самая ранняя таская клиента, который обрабатывался раньше всех по истории.
Величина истории ограничена 100, если надо можно вынести это в параметр пула.

package net.sf.brunneng;


import java.util.*;

public class ThreadPool {

   private class ClientTask {
      final int clientId;
      final Runnable runnable;

      public ClientTask(int clientId, Runnable runnable) {
         this.clientId = clientId;
         this.runnable = runnable;
      }
   }

   private static final int MAX_EXECUTED_CLIENTS_HISTORY = 100;

   private boolean closed;
   private final List<ClientTask> tasks = Collections.synchronizedList(new ArrayList<ClientTask>());

   private List<Thread> threads = new ArrayList<Thread>();
   private Set<Integer> executingClients = Collections.synchronizedSet(new HashSet<Integer>());

   private LinkedList<Integer> executedClientsHistory = new LinkedList<Integer>();

   public List<ClientTask> getTasks() {
      return tasks;
   }

   public ThreadPool(int threadsCount) {
      for (int i = 0; i < threadsCount; ++i) {
         Thread thread = createThread();
         threads.add(thread);
         thread.start();
      }
   }

   private ClientTask getNextClientTask() {
      ClientTask res = null;

      synchronized (tasks) {

         Map<Integer, Integer> clientIdToEarliestTaskInstanceMap = new HashMap<Integer, Integer>();
         for (int i = 0; i < tasks.size(); i++) {
            ClientTask task = tasks.get(i);
            if (!executingClients.contains(task.clientId) &&
                    !clientIdToEarliestTaskInstanceMap.containsKey(task.clientId)) {
               clientIdToEarliestTaskInstanceMap.put(task.clientId, i);
            }
         }

         final Map<Integer, Integer> clientIdMinStepsBack = new HashMap<Integer, Integer>();
         Iterator<Integer> iterator = executedClientsHistory.descendingIterator();
         int i = 1;
         while (iterator.hasNext()) {
            Integer clientId = iterator.next();
            if (!clientIdMinStepsBack.containsKey(clientId)) {
               clientIdMinStepsBack.put(clientId, i);
            }
            i++;
         }

         int maxStepBack = -1;
         Integer bestEarliestTaskIndex = null;
         for (Integer clientId : clientIdToEarliestTaskInstanceMap.keySet()) {
            Integer stepsBack = clientIdMinStepsBack.get(clientId);
            if (stepsBack == null) { // no occurence in history
               bestEarliestTaskIndex = clientIdToEarliestTaskInstanceMap.get(clientId);
               break;
            }
            else if (stepsBack > maxStepBack) {
               maxStepBack = stepsBack;
               bestEarliestTaskIndex = clientIdToEarliestTaskInstanceMap.get(clientId);
            }
         }

         if (bestEarliestTaskIndex != null) {
            res = tasks.remove(bestEarliestTaskIndex.intValue());
            executingClients.add(res.clientId);

            executedClientsHistory.add(res.clientId);
            if (executedClientsHistory.size() > MAX_EXECUTED_CLIENTS_HISTORY) {
               executedClientsHistory.removeFirst();
            }
         }
      }

      return res;
   }

   private Thread createThread() {
      return new Thread(new Runnable() {
         @Override
         public void run() {
            boolean finish = false;

            while (!finish && !closed) {
               try {
                  ClientTask task = getNextClientTask();

                  if (task == null) {
                     synchronized (tasks) {
                        tasks.wait();
                     }
                     continue;
                  }

                  task.runnable.run();
                  executingClients.remove(task.clientId);

               } catch (InterruptedException e) {
                  e.printStackTrace();
                  finish = true;
               }
            }
         }
      });
   }

   public void addTask(int clientId, Runnable task) {
      synchronized (tasks) {
         tasks.add(new ClientTask(clientId, task));

         if (!executingClients.contains(clientId)) {
            tasks.notify();
         }
      }
   }

   public void close() {
      closed = true;

      synchronized (tasks) {
         tasks.notifyAll();
      }

      for (Thread thread : threads) {
         try {
            thread.join();
         } catch (InterruptedException ignored) {

         }
      }
   }
}