public class TesPool {

    private final ProcessingThread[] threads;
    private volatile boolean isShutdown = false;
    private final AtomicInteger pendingTasks = new AtomicInteger();
    private final ConcurrentMap<Object, TaskStuff> cm = new ConcurrentHashMap<Object, TaskStuff>();

    private final TaskStuff ZEROCOUNTTASKSTUFF = new TaskStuff(0, null);

    public TesPool(int size) {
        ProcessingThread[] pool = new ProcessingThread[size];
        for(int i = 0; i < size; i++) {
            pool[i] = new ProcessingThread();
            pool[i].setDaemon(true);
        }
        threads = pool;
    }

    public void start() {
        for(Thread thread : threads) {
            thread.start();
        }
    }

    public boolean shutdown() {
        isShutdown = true;
        while(true) {
            int shut = 0;
            for(Thread th : threads) {
                if(!th.isAlive()) {
                    shut++;
                }
            }
            if(shut == threads.length) {
                return true;
            }
            LockSupport.parkNanos(300);
        }
    }

    public void submit(Actionee action) throws RejectedExecutionException {
        pendingTasks.incrementAndGet();
        if(isShutdown) {
            pendingTasks.decrementAndGet();
            throw new RejectedExecutionException("shutdown!");
        }

        TaskStuff newTs = new TaskStuff(0, null);
        TaskStuff currentTs = null;
        Random rnd = new Random();
        do {
            newTs.setCount(1);
            newTs.setThread(threads[Math.abs(rnd.nextInt()) % threads.length]);
            currentTs = cm.putIfAbsent(action.getKey(), newTs);
            if(currentTs == null) {
                break;
            }
            newTs.setCount(currentTs.getCount() + 1);
            newTs.setThread(currentTs.getThread());
        } while(!cm.replace(action.getKey(), currentTs, newTs));
        newTs.getThread().submitToThread(action);
    }

    public class ProcessingThread extends Thread {

        private final BlockingQueue<Actionee> blockingQueue = new LinkedBlockingQueue<Actionee>();

        @Override
        public void run() {
            while(!isShutdown || !blockingQueue.isEmpty() || pendingTasks.get() != 0) {
                while(true) {
                    Actionee actionee = blockingQueue.poll();
                    if(actionee == null) {
                        break;
                    }
                    try {
                        actionee.run();
                    } catch(Exception ex) {
                        ex.printStackTrace();
                    } finally {
                        TaskStuff newTs = new TaskStuff(0, this);
                        while(true) {
                            TaskStuff ts = cm.get(actionee.getKey());
                            newTs.setCount(ts.getCount() - 1);
                            if(cm.replace(actionee.getKey(), ts, newTs)) {
                                break;
                            }
                        }
                        cm.remove(actionee.getKey(), ZEROCOUNTTASKSTUFF);
                    }
                }
                LockSupport.parkNanos(300);
            }
        }

        public void submitToThread(Actionee actionee) {
            try {
                blockingQueue.add(actionee);
            } finally {
                pendingTasks.decrementAndGet();
            }
        }
    }

    private static class TaskStuff {
        private int count;
        private ProcessingThread thread;

        public TaskStuff(int count, ProcessingThread thread) {
            this.count = count;
            this.thread = thread;
        }

        public int getCount() {
            return count;
        }

        public void setCount(int count) {
            this.count = count;
        }

        public ProcessingThread getThread() {
            return thread;
        }

        public void setThread(ProcessingThread thread) {
            this.thread = thread;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;

            TaskStuff taskStuff = (TaskStuff) o;

            return count == taskStuff.count;
        }
    }

    public interface Actionee extends Runnable {
        Object getKey();
    }

    // тесты
    public static void main(String[] args) throws Exception {
        final TesPool tp = new TesPool(256);
        tp.start();
        for(int i = 0; i < 5; i++) {
            long begin = System.currentTimeMillis();
            final CountDownLatch cdlDone = new CountDownLatch(100000);
            new Thread() {
                @Override
                public void run() {
                    final Random rdm = new Random();
                    for (int i = 0; i < 100000; i++) {
                        tp.submit(new Actionee() {

                            private final int key;

                            {
                                key = rdm.nextInt();
                            }

                            @Override
                            public Object getKey() {
                                return key;
                            }

                            @Override
                            public void run() {
                                try {
                                    Thread.sleep(1);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                } finally {
                                    cdlDone.countDown();
                                }
                            }
                        });
                    }
                }
            }.start();
            cdlDone.await();
            /*Map<ProcessingThread, Integer> thSet = new HashMap<ProcessingThread, Integer>();
            for(TaskStuff taskStuff : tp.cm.values()) {
                if(thSet.containsKey(taskStuff.getThread())) {
                    thSet.put(taskStuff.getThread(), thSet.get(taskStuff.getThread()) + taskStuff.getCount());
                } else {
                    thSet.put(taskStuff.getThread(), taskStuff.getCount());
                }
            }
            int total = 0;
            for(Map.Entry<ProcessingThread, Integer> entry : thSet.entrySet()) {
                total += entry.getValue();
                System.out.println("Thread : " + entry.getKey() + " count: " + entry.getValue());
            }
            System.out.println("Total threads used: " + thSet.size() + " total tasks: " + total);*/
            System.out.println("TotalDone: 100000"  + ", took: " + (System.currentTimeMillis() - begin) + " ms");
        }
        tp.shutdown();
    }

}