3
\$\begingroup\$

I have a service that is subscribed to a specific channel of ActiveMQ.

Every time a message arrives the service has to start a "job" (submit instances of Callables).

Since there will be thousands of messages, in order to limit the throughput, I was thinking about implementing a custom ThreadPoolExecutor with these features:

  • a human friendly name
  • a bounded queue
  • a listener / callback to notify the service about accepted/rejected/terminated threads.

Besides since will be great to log the pool name with the accepted/rejected/terminated callable name (specially during the debug) I hacked both the ThreadPoolExecutor and the ThreadFactory.

Everything works fine (right now I cannot test under the thousands messages) but I'll be pleased to be warned by someone with more experience in this kind of problems.

And if there are better and known patterns/solutions, please let me know (but I'd like to avoid huge frameworks with huge dependencies whit other huge dependencies :-) )

Thank you very much,

Luca


LabeledThreadPoolFactory

This is a custom ThreadPoolFactory that simply give the specified name to the pool (useful for debug and logging).

public class LabeledThreadPoolFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);

    private final ThreadGroup group;
    private final AtomicInteger threadNumber;
    private final String namePrefix;


    public LabeledThreadPoolFactory(final String name) {
        SecurityManager s = System.getSecurityManager();
        this.group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        this.threadNumber = new AtomicInteger(1);
        this.namePrefix = "pool-" + this.poolNumber.getAndIncrement() + "-" + name + "-thread-";
    }


    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(),0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

MonitoredThreadPool

This is the ThreadPoolExecutor that notify when a Callable is accepted/rejected/terminated (we could also expose the execution time eventually).

public class MonitoredThreadPool extends ThreadPoolExecutor {
    public interface MonitoredThreadPoolListener {
        void onThreadAccepted(Runnable r);
        void onThreadRejected(Runnable r);
        void onThreadTerminated(Runnable r);
    }

    private static final int DEFAULT_POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(), 2);

    private final Map<Runnable, Boolean> inProgress = new ConcurrentHashMap<>();
    private final ThreadLocal<Long> startTime = new ThreadLocal<>();
    private long totalTime;
    private int totalTasks;
    private MonitoredThreadPoolListener monitoredThreadPoolListener;


    public MonitoredThreadPool(final String threadPoolName) {
        this(threadPoolName, DEFAULT_POOL_SIZE, DEFAULT_POOL_SIZE,
                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(10));
    }

    public MonitoredThreadPool(final String threadPoolName,
                               final int corePoolSize, final int maximumPoolSize,
                               final long keepAliveTime, final TimeUnit unit,
                               final BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                new LabeledThreadPoolFactory(threadPoolName));
        this.monitoredThreadPoolListener = null;
    }

    public MonitoredThreadPool(int corePoolSize,
                               int maximumPoolSize,
                               long keepAliveTime,
                               TimeUnit unit,
                               BlockingQueue<Runnable> workQueue,
                               ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.monitoredThreadPoolListener = null;
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable) {
            @Override
            public String toString() {
                return callable.toString();
            }
        };
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        inProgress.put(r, Boolean.TRUE);
        startTime.set(System.currentTimeMillis());

        if (this.monitoredThreadPoolListener != null)
            this.monitoredThreadPoolListener.onThreadAccepted(r);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        long time = System.currentTimeMillis() - startTime.get();
        synchronized (this) {
            totalTime += time;
            ++totalTasks;
        }
        inProgress.remove(r);
        super.afterExecute(r, t);
        if (this.monitoredThreadPoolListener != null)
            this.monitoredThreadPoolListener.onThreadTerminated(r);
    }

    public void setMonitoredThreadPoolListener(MonitoredThreadPoolListener listener) {
        this.monitoredThreadPoolListener = listener;
        if (listener != null) {
            setRejectedExecutionHandler(new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    if (MonitoredThreadPool.this.monitoredThreadPoolListener != null)
                        MonitoredThreadPool.this.monitoredThreadPoolListener.onThreadRejected(r);
                    //.TODO passare anche qualche statistica?
                }
            });
        }
    }

    public Set<Runnable> getInProgressTasks() {
        return Collections.unmodifiableSet(inProgress.keySet());
    }

    public synchronized int getTotalTasks() {
        return totalTasks;
    }

    public synchronized double getAverageTaskTime() {
        return (totalTasks == 0) ? 0 : totalTime / totalTasks;
    }
}

Quick and dirty sample usage

This is a very quick and dirty samle about the usage f the two previous classes.

public class SampleService implements MonitoredThreadPool.MonitoredThreadPoolListener {

    private final ThreadPoolExecutor executor;

    public SampleService() {
        MonitoredThreadPool executor = new MonitoredThreadPool("bazinga");
        executor.setMonitoredThreadPoolListener(this);

        this.executor = executor;
    }

    public void test() {
        for (int i = 0; i < 20; i++) {
            String name = "callable_" + i;
            this.executor.submit(newCallable(name, 5));
        }
    }

    public void shutdown() {
        //.Shutdown the Thread Pool Executor
        if (this.executor != null) {
            try {
                this.executor.shutdown();
                this.executor.awaitTermination(1, TimeUnit.MINUTES);
            } catch (InterruptedException ignore) {
                //LOG.warn("tasks interrupted", e);
            } finally {
                executor.shutdownNow();
            }
        }
    }

    private Callable<String> newCallable(final String text, final int seconds) {
        return new Callable<String>() {
            @Override
            public String call() throws Exception {
                System.out.printf("Hi! I\'m %s now I\'ll sleep for %d seconds\n", text, seconds);
                try {
                    Thread.sleep(seconds*1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return null;
            }

            @Override
            public String toString() {
                return text;
            }
        };
    }

    @Override
    public void onThreadAccepted(Runnable r) {
        System.out.println("welcome to the pool: " + r.toString());
    }

    @Override
    public void onThreadRejected(Runnable r) {
        System.err.println("sorry, all booked: " + r.toString());
    }

    @Override
    public void onThreadTerminated(Runnable r) {
        System.out.println("hope you enjoyed the pool: " + r.toString());
    }


    public static void main(String[] args) {
        SampleService sampleService = new SampleService();
        sampleService.test();

        try {
            Thread.sleep(1000*30);
            //shut down the pool
            sampleService.shutdown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.