4
\$\begingroup\$

Despite my experience with Java, I am not well-versed with concurrency. I wrote this thread pool that has a queue of tasks. Please ignore any signature choices, such as overuse of final. Thank you for your feedback!

public final class ThreadPool {

  // Logging
  //--------------------------------------------------

  private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPool.class);

  // Constructors
  //--------------------------------------------------

  public ThreadPool(final int size, final boolean lazy) {
    super();

    this.size = size;
    this.workers = new Worker[this.size];
    this.queue = new LinkedBlockingQueue<>();

    if(!lazy) createAndStartWorkers();
  }

  // Fields
  //--------------------------------------------------

  private volatile boolean initialized = false;

  private final int size;

  private final Worker[] workers;

  private final BlockingQueue<Runnable> queue;

  // Methods
  //--------------------------------------------------

  public final void schedule(final Runnable task) {
    if(!initialized) createAndStartWorkers();

    synchronized(queue) {
      queue.add(task);
      queue.notify();
    }
  }

  public final void shutdown() {
    for(int i = 0; i < size; i++) {
      workers[i] = null;
    }
  }

  public final int getSize() {
    return size;
  }

  private void createAndStartWorkers() {
    for(int i = 0; i < size; i++) {
      workers[i] = new Worker();
      workers[i].start();
    }

    initialized = true;
  }

  // Nested
  //--------------------------------------------------

  /**
   * Worker thread.
   *
   * @author Oliver Yasuna
   * @since 2.0.0
   */
  private final class Worker extends Thread {

    // Constructors
    //--------------------------------------------------

    private Worker() {
      super();
    }

    // Overrides
    //--------------------------------------------------

    // Runnable
    //

    @Override
    public final void run() {
      Runnable task;

      while(true) {
        synchronized(queue) {
          while(queue.isEmpty()) {
            try {
              queue.wait();
            } catch(final InterruptedException e) {
              LOGGER.warn("An error occurred while waiting for the queue.", e);
            }
          }

          task = queue.poll();
        }

        try {
          task.run();
        } catch(final RuntimeException e) {
          LOGGER.warn("Thread pool was interrupted due to an exception.", e);
        }
      }
    }

  }

}
\$\endgroup\$
4
  • 1
    \$\begingroup\$ You may notice a trend in my coding style. I don't like implicit code. For example, the class is final, yet I still make all public methods final. Same with constructors, which implicitly call super. \$\endgroup\$
    – Oliver
    Jun 29, 2022 at 14:17
  • \$\begingroup\$ Well didn't think of extending object behind the scenes. \$\endgroup\$
    – Heslacher
    Jun 29, 2022 at 14:29
  • \$\begingroup\$ What is the purpose of this code? Is it homework, personal project, professional? \$\endgroup\$
    – Eric Stein
    Jun 29, 2022 at 18:30
  • \$\begingroup\$ @EricStein Personal to learn more about threading in Java. I've always let frameworks (e.g., Spring) handle threading for me. I decided I want to learn how these work, so I'm starting simple. \$\endgroup\$
    – Oliver
    Jun 30, 2022 at 15:24

1 Answer 1

2
\$\begingroup\$

I am assuming this is not intended for serious use, as there’s no good reason not to use the tools already provided by Java for pooling threads made available in Executors. As I am not an expert in concurrent programming in Java, I’m sure I’ve missed at least one important thing. Strongly consider picking up the excellent Java Concurrency in Practice. It is somewhat dated, but the core of the concurrency library was written before the book. You can also look at the implementation of ThreadPoolExecutor, which will show you all the things the API writers felt were necessary.

All of the comments which are used to visually separate “sections” are noise, and make it harder to actually read the code.

In idiomatic java

  • instance variables are declared before constructors, not after.
  • There is whitespace between control flow operators (for, while,if, catch) and open parens. This visually distinguishes them from method calls, where the paren is adjacent.

queue can and should be assigned when it is declared.

size does not need to be tracked separately, as it can always be determined from the length of the workers array.

super() in the constructor is noise. The no-argument constructor of Object does nothing.

Validation should be added for size in the constructor. A size of zero will create a thread pool which accepts tasks but never executes them.

I do not see any value in waiting to start the workers. Sleeping threads do not typically consume significant resources. The code complexity is almost certainly not worth the marginal system resource gain.

Even when optional, curly braces should always be used. They improve readability and eliminate a common class of programming error.

Synchronizing on queue is not leveraging a primary feature of LinkedBlockingQueue. The take() method will block until an item is available. Each worker thread should call take() instead of waiting to be notified.

Shutdown doesn’t. Making the pointers to the worker threads null doesn’t affect the threads themselves at all. They will happily continue running and acquiring tasks from the queue. The threads need to be interrupted. The queue needs to be drained so the threads don’t grab any new tasks. The schedule method needs to check if the thread pool is running and disallow adding new tasks.

Javadoc that merely says “Worker Thread” is meaningless and can be removed. If documentation doesn’t provide value, it’s just making it harder to read the important bits.

Worker should not extends Thread. You have not created a new, generally-usable type of Thread. You have a specific block of code you want a thread to run. Worker should implements Runnable.

Your personal preferences aside, the constructor is noise and should be removed.

As mentioned above, using take() in run() removes a lot of noise from the method.

An InterruptedException is not “An error occurred”. It’s “Somebody asked me to stop what I’m doing”. The worker should stop running when it sees an InterruptedException. It is most likely an info(), not a warn().

A RuntimeException is not “Thread pool was interrupted”. It’s “The current task threw an exception”. It should be an error(), not a warn().

If you made all these changes, your code might look more like:

public final class ThreadPool {


    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPool.class);

    private final Thread[] workers;
    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
    private volatile boolean running = true;

    public ThreadPool(final int size) {
        this.workers = new Thread[size];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new Thread(new Worker());
            workers[i].start();
        }
    }

    public final void schedule(final Runnable task) {
        if (running) {
            queue.add(task);
        } else {
            // TODO: handle this case. Log, throw, etc.
        }
    }

    public final void shutdown() {
        running = false;
        queue.clear();
        for (int i = 0; i < workers.length; i++) {
            workers[i].interrupt();
        }
    }

    public final int getSize() {
        return workers.length;
    }

    /**
     * @author Oliver Yasuna
     * @since 2.0.0
     */
    private final class Worker implements Runnable {

        @Override
        public final void run() {
            while (true) {
                try {
                    queue.take().run();
                } catch (final InterruptedException ie) {
                    LOGGER.info("Thread " + Thread.currentThread().getName() + " interrupted.", ie);
                    break;
                } catch (final RuntimeException e) {
                    LOGGER.error("Worker task failed due to an exception.", e);
                }
            }
        }

    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.