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

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

GT>>Если кто шарит в java, прошу поревьювить. Я сам с ручной многопоточностью редко работал, так что могут быть косяки..

EP>Представь ситуацию, когда в первой половине tasks у нас задачи от первого клиента, а во второй — от второго, и всего один рабочий поток. Он будет выполнять все задания по списку, то есть сначала все от первого клиента, хотя по заданию должен чередовать.

Согласен.. Поправил.

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 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>());

   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) {
         for (int i = 0; i < tasks.size(); i++) {
            ClientTask task = tasks.get(i);
            if (!executingClients.contains(task.clientId)) {
               res = task;
               executingClients.add(task.clientId);
               tasks.remove(i);
               break;
            }
         }
      }

      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) {
         int nextIndex = findBestIndexToInsert(clientId);
         tasks.add(nextIndex, new ClientTask(clientId, task));

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

   private int findBestIndexToInsert(int clientId) {
      int minIndex = 0;

      for (int i = 0; i < tasks.size(); ++i) {
         ClientTask task = tasks.get(i);
         if (task.clientId == clientId) {
            minIndex = i + 1;
         }
      }

      int bestIndex = minIndex;

      for (int i = minIndex; i < tasks.size(); ++i) {
         ClientTask task = tasks.get(i);
         if (clientId != task.clientId) {
            bestIndex = i + 1;
            break;
         }
      }

      return bestIndex;
   }

   public void close() {
      closed = true;

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

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

         }
      }
   }
}

Идея метода findBestIndexToInsert такая: вначале ищем minIndex — индекс таски следующей за последней таской с заданным clientId. Так обеспечиваем что все таски от одного клиента идут подряд. Далее ищем первую таску, у которой клиент будет отличаться от clientId. Вставлять новую таску будем сразу за ней..

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

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

GT>>Если кто шарит в java, прошу поревьювить. Я сам с ручной многопоточностью редко работал, так что могут быть косяки..

EP>Представь ситуацию, когда в первой половине tasks у нас задачи от первого клиента, а во второй — от второго, и всего один рабочий поток. Он будет выполнять все задания по списку, то есть сначала все от первого клиента, хотя по заданию должен чередовать.

Согласен.. Поправил.

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 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>());

   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) {
         for (int i = 0; i < tasks.size(); i++) {
            ClientTask task = tasks.get(i);
            if (!executingClients.contains(task.clientId)) {
               res = task;
               executingClients.add(task.clientId);
               tasks.remove(i);
               break;
            }
         }
      }

      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) {
         int nextIndex = findBestIndexToInsert(clientId);
         tasks.add(nextIndex, new ClientTask(clientId, task));

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

   private int findBestIndexToInsert(int clientId) {
      int minIndex = 0;

      for (int i = 0; i < tasks.size(); ++i) {
         ClientTask task = tasks.get(i);
         if (task.clientId == clientId) {
            minIndex = i + 1;
         }
      }

      return minIndex < tasks.size() ? minIndex + 1 : minIndex;
   }

   public void close() {
      closed = true;

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

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

         }
      }
   }
}