How to make an optimized scenario for Java concurrent application?

Question

I've been looking for similar questions like these, but for my requirements I think I need something special, I will explain it in detail.

First of all I need to migrate a system which used to work this way:

1. A class called ServerPool(Thread) initializes with a main class.
2. This ServerPool creates a queue to receive sockets and a vector to manage worker threads (S.

So in code for pool, I have the following:

public class ServerPool extends Thread {
    private LinkedBlockingQueue<SearchQuery> workQueue; //SearchQuery is a Class I defined which can handle two type of processes (for sockets and for single Strings)
    private Vector<SearchThread> workers;
    private final int NTHREADS = 10;
    private int typeOfQuery;

    public ServerPool() {
    workers = new Vector<SearchThread>(NUM_THREAD);
    workQueue = new LinkedBlockingQueue<SearchQuery>();
    this.typeOfQuery = typeOfQuery;
    SearchThread search = new SearchThread(workQueue);
    search.start();
    workers.add(search);
}

public void run() {
    while(true){
    SearchQuery client = null;

    if (typeOfQuery == 1) {
        client = new SocketQuery(....);
        } else if (typeOfQuery == 2) {
        client = new StringQuery(...);
        }
        workQueue.put(client);
}
}

For the SearchThread which executes the process:

public class SearchThread extends Thread {
    private LinkedBlockingQueue<SearchQuery> workQueue = null;
    private SearchQuery request = null;

    public SearchThread(LinkedBlockingQueue<SearchQuery> workSource) {
        workQueue = workSource;
    }   

    public void run() {
        request = workQueue.take();
        //Here I process the request
        //And use a PrintWriter to give a "response"            
    }

}

This used to work using telnet with sockets, but now I've been asked to convert it into a Web Service, so as Web Service it is supposed to return a value, so I think of using Callable, Future and Thread Pools, but I can't replicate exactly the same behavior, I tried implementing this:

public class NewServerPool {

    private final int NTHREADS = 10;
    private ExecutorService executor;
    private LinkedBlockingQueue<SearchQuery> workQueue;
    private Vector<Future<String>> futures;
    private boolean end = true;

    public NewServerPool(int port, SearchQuery typeOfQuery) {
        executor = Executors.newFixedThreadPool(NTHREADS);
        workQueue = new LinkedBlockingQueue<SearchQuery>();
        futures = new Vector<Future<String>>();

    }


}

And for the Search Thread that now it is a Callable

public class NewSearchThread implements Callable<String>{

    private SearchQuery searchQuery;

    public NewSearchThread(SearchQuery searchQuery) {
        this.searchQuery = searchQuery;
    }

    @Override
    public String call() throws Exception {
        String xmlResponse = null;

        if (searchQuery == null) {
            throw new InvalidSearchQueryException("The search query is not valid or has null value: " + searchQuery);
        }

        if (searchQuery instanceof SocketTimed) {
            System.out.println("It is socket timed query type");
        } else if (searchQuery instanceof WebServiceQuery) {
            System.out.println("It is a web service query type");
        }

        xmlResponse = searchQuery.manageResponse();

        return xmlResponse;
    }

So I've got stucked in server pool, asumming my WebService will invoke a new instance of Server Pool (NewServerPool) in this case, how could I continue with this? Please I will be really grateful if somebody can help me. Thanks in advance, best regards.

Answer 1

If you just want to "launch" the thread pool and return a value, then it doesn't sound like there's any reason to have your NewServerThreadPool extend Thread (but without knowing the full specification of what you are trying to achieve I'm not 100% sure). What type of value is your launch method supposed to return? boolean? String? int? You could instead try something like this:

public class NewServerPool {

    private final int NTHREADS = 10;
    private ExecutorService executor;
    private Vector<Future<String>> futures;

    public NewServerPool(int port, SearchQuery typeOfQuery) {
        futures = new Vector<Future<String>>();
    }

    public boolean launchThreadPool() {
        executor = Executors.newFixedThreadPool(NTHREADS);
        return true;
    }

    public void submitToThreadPoolForProcessing(SearchQuery client) {
        futures.add(executor.submit(new NewSearchThread(client)));
    }

    public Vector<Future<String>> getFutures() {
        return futures;
    }
}

Note that in the above, the single-line-contents of the launchThreadPool() method could just as easily be part of the constructor (as it was in the previous post), but breaking it out into its own method allows you to return value "after launching the thread pool. As shown it will return a boolean value (always will return true), but you can of course change the method to return whatever type your specification calls for.

Answer 2

A couple of things:

First off, your original ServerPool class is flawed, in that it only ever instantiates 1 instance of SearchThread. I think you meant it to start NTHREADS (10) SearchThreads.

Next, it looks like you've changed the approach of NewSearchThread slightly from SearchThread - in that the constructor for NewSearchThread takes a SearchQuery argument, whereas the SearchThread takes a SearchQuery off of the BlockingQueue.

And finally your NewServerPool class differs in its approach from ServerPool, in that ServerPool's run() method continuously places new SearchQuerys into the BlockingQueue. In contrast, NewServerPool's constructor takes a single SearchQuery and does nothing with it.

How about something like this to get you started:

public class NewServerPool extends Thread {

    private final int NTHREADS = 10;
    private ExecutorService executor;
    private Vector<Future<String>> futures;

    public NewServerPool(int port, SearchQuery typeOfQuery) {
        executor = Executors.newFixedThreadPool(NTHREADS);
        futures = new Vector<Future<String>>();
    }

    public void run() {
        while(true){
            SearchQuery client = null;

            if (typeOfQuery == 1) {
                client = new SocketQuery(....);
            } else if (typeOfQuery == 2) {
                client = new StringQuery(...);
            }
            futures.add(executor.submit(new NewSearchThread(client)));
        }
    }
}

Note that I say "to get you started"... as the above still needs some additions such as proper exiting of the run() method when it's time to stop fielding requests (but that's another topic).