1
\$\begingroup\$

I am studying multi-threading and have tried to implement a thread pool. Please provide feedback, mostly from multi-threading point of view. I have implemented two pools, one for adding tasks in queue and the other for taking and executing the requests.

PoolManager.java

package com.learn.threading.threadpool;
public class PoolManager {
    ServiceThreadPool servicePool;
    RequestThreadPool requestPool;
    public static void main(String[] args) {
        PoolManager poolManager = new PoolManager();
        poolManager.start();
        try{
            Thread.sleep(4000);
        }catch(InterruptedException ex){
             ex.printStackTrace();
        }finally {
             poolManager.stop();
        }
    }

    public void start(){
        Queue queue = new Queue(10);
        servicePool = new ServiceThreadPool(10,queue);
        requestPool = new RequestThreadPool(1000,queue);
        servicePool.start();
        requestPool.start();
    }   

    public void stop(){
        requestPool.stop();
        servicePool.stop();
    }
}

ServiceThreadPool.java

package com.learn.threading.threadpool;

public class ServiceThreadPool {

    private int numOfThreads;
    private Processor[] threads;

    public ServiceThreadPool(int numOFthreads, Queue queue) {
        this.numOfThreads = numOFthreads;
        threads = new Processor[numOFthreads];
        for(int i = 0; i < numOFthreads; i++){
            threads[i] = new Processor(queue, "Processor_"+i+" ");
        }
    }

    public void start(){
        for(int i = 0; i < numOfThreads; i++){
            threads[i].start();
        }
    }

    public void stop(){
        for(int i = 0; i < numOfThreads; i++){
            threads[i].interrupt();
        }
    }
}

RequestThreadPool.java

package com.learn.threading.threadpool;

public class RequestThreadPool {
    private int numOfThreads;
    private Request[] threads;

    public RequestThreadPool(int numOFthreads, Queue queue) {
        this.numOfThreads = numOFthreads;
        threads = new Request[numOFthreads];
        for(int i = 0; i < numOFthreads; i++){
            threads[i] = new Request(queue,"Request_"+i+" ");
        }
    }

    public void start(){
        for(int i = 0; i < numOfThreads; i++){
            threads[i].start();
        }
    }

    public void stop(){
        for(int i = 0; i < numOfThreads; i++){
            threads[i].interrupt();
        }
    }
}

Request.java

package com.learn.threading.threadpool;

public class Request extends Thread {
    private Queue queue;
    private String name;
    volatile private static int unique = 1;
    public Request(Queue queue, String name) {
        this.queue = queue;
        this.name = name;
    }

    public void run() {
        while (!isInterrupted()) {
            try {
                Task task = new Task(unique++);
                synchronized (queue) {
                    while (queue.isFull()) {
                        queue.wait();
                    }
                    queue.add(task);
                    System.out.println(task +" added in Queue by "+name);
                    queue.notifyAll();
                }                       
            } catch (InterruptedException ex) {
                break;
            }
        }
    }
}

Processor.java

package com.learn.threading.threadpool;

public class Processor extends Thread {
    private volatile Queue queue;
    private String name;
    public Processor(Queue queue, String name) {
        this.queue = queue;
        this.name = name;
    }

    @Override
    public void run() {
        while (!isInterrupted()) {
            try {
                Runnable task;
                synchronized (queue) {
                    while (queue.isEmpty()) {
                        queue.wait();
                    }
                    task = queue.remove();
                    System.out.println(task +" taken from Queue by "+name);
                    queue.notifyAll();
                }
                task.run();
            } catch (InterruptedException ex) {
                break;
            }
        }
    }
}

Queue.java

package com.learn.threading.threadpool;

public class Queue  {
    Runnable[] requests;
    int maxSize;
    int size;

    public Queue(int maxSize){
        this.maxSize = maxSize;
        requests = new Task[maxSize];
    }

    public void add(Runnable task){
        requests[size++] = task;
    }

    public Runnable remove(){
        Runnable task = requests[size-1];
        requests[size-1] = null;
        size--;
        return task;
    }

    public boolean isFull(){
       return size == maxSize;
    }

    public boolean isEmpty(){
       return size == 0;
    }
}

Task.java

package com.learn.threading.threadpool;

public class Task implements Runnable {
    private int input;

    public Task(int input){
        this.input = input;
    }

    public void run() {
        int result = 0;
        int temp = input;
        while(temp > 1){
            result = result+temp;
            temp = temp-1;
        }
        System.out.println("sum of 1st "+input+" numbers is "+result);
    }

    public String toString(){
        return "Task "+input;
    }
}
\$\endgroup\$

1 Answer 1

3
\$\begingroup\$
            Queue queue = new Queue(10);
            servicePool = new ServiceThreadPool(10,queue);
            requestPool = new RequestThreadPool(1000,queue);

You should replace those numbers with constants. Reading that I'm not sure -- is the ServiceThreadPool the same size as the queue? Or is that a coincidence? With constants, it would be obvious.

An alternative would be to call this method with each value as a parameter.

    public ServiceThreadPool(int numOFthreads, Queue queue) {
            this.numOfThreads = numOFthreads;

This isn't necessary. You could either say

    public ServiceThreadPool(int numOFthreads, Queue queue) {
            numOfThreads = numOFthreads;

or

    public ServiceThreadPool(int numOfThreads, Queue queue) {
            this.numOfThreads = numOfThreads;

It's not necessary to do both.

I'd recommend the latter, as using capitalization to differentiate variables is prone to misunderstandings. Not to mention against the naming standard.

            for(int i = 0; i < numOfThreads; i++){
                    threads[i].start();
            }

This is a great place to use the for each form.

            for (Request thread : threads) {
                    thread.start();
            }

It's just simpler.

                    try {
                            Runnable task;
                            synchronized (queue) {
                                    while (queue.isEmpty()) {
                                            queue.wait();
                                    }
                                    task = queue.remove();
                                    System.out.println(task +" taken from Queue by "+name);
                                    queue.notifyAll();
                            }
                            task.run();

The normal standard for Java is four-column indents. With eight-column, it's recommended to not allow more than two levels of indent for the reason shown here -- the indent takes up too much of the screen. To get rid of the extra levels of indent, you can put the code shown here (except the try) into another method.

                    } catch (InterruptedException ex) {
                            break;
                    }

Why break here? You could just return. Either will have the same ultimate effect, but return is more obvious about what happens.

                    try {
                            runTask();
                    } catch (InterruptedException ex) {
                            return;
                    }

And of course you'd have to implement runTask.

    public Runnable remove(){
            Runnable task = requests[size-1];
            requests[size-1] = null;
            size--;
            return task;
    }

OK. So you subtract one from size. Then do it again. Then again and this time you store it. Wouldn't it be simpler to just

    public Runnable remove(){
            Runnable task = requests[--size];
            requests[size] = null;
            return task;
    }

That saves two subtractions. Note that if you want, you can move the decrement operation to its own line at the beginning of the method. Some find that clearer.

Note also that this does not implement a queue. This is a stack. A queue would always add to the end and remove from the beginning so that the first added is the first out.

            int result = 0;
            int temp = input;
            while(temp > 1){
                    result = result+temp;
                    temp = temp-1;
            }
            System.out.println("sum of 1st "+input+" numbers is "+result);

Why not just say

            int result = 0;
            for (int i = 2; i <= input; ++i) {
                    result += i;
            }
            System.out.println("sum of 1st "+input+" numbers excepting 1 is "+result);

Note that I changed the println to more accurately reflect what the code was doing. Alternately, you could change the for loop to

            for (int i = 1; i <= input; ++i) {

To match the println instead. Or if you'd prefer

            for (int i = input; i > 0; --i) {

And I prefer i to temp as a name for a loop variable. Or replace the whole method with just

            System.out.println("sum of 1st "+input+" numbers is "+ (input * (input + 1) / 2));

Which will give the same result with only three calculations. Note that the original does two comparisons and an addition even if input is equal to 1. Since comparisons are often implemented as a subtraction and a comparison to 0, this means three calculations is the minimum for the original. It grows by two when input grows by one.

Of course, you may be doing the sum slowly deliberately. In which case optimizing it would be counter-productive.

\$\endgroup\$

Your Answer

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

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