java - 如何为Java并发应用制定优化场景？

Question

我一直在寻找类似的问题，但对于我的要求，我认为我需要一些特别的东西，我会详细解释它。

首先，我需要迁移一个以前以这种方式工作的系统:

1. 名为 ServerPool(Thread) 的类使用主类进行初始化。
2. 此 ServerPool 创建一个队列来接收套接字和一个 vector 来管理工作线程 (S.

因此，在池代码中，我有以下内容:

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);
}
}

对于执行该进程的SearchThread:

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"            
    }

}

这曾经使用带套接字的 telnet 来工作，但现在我被要求将其转换为 Web 服务，因此作为 Web 服务它应该返回一个值，所以我想到使用 Callable、Future 和线程池，但我无法复制完全相同的行为，我尝试实现此:

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>>();

    }


}

对于搜索线程来说，它现在是一个可调用的

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;
    }

所以我被困在服务器池中，假设在这种情况下我的 WebService 将调用服务器池的新实例 (NewServerPool)，我该如何继续？如果有人能帮助我，我将非常感激。 预先感谢，最诚挚的问候。

Answer 1

有几件事:

首先，您原来的 ServerPool 类是有缺陷的，因为它只实例化了 1 个 SearchThread 实例。我认为您的意思是启动 NTHREADS (10) SearchThreads。

接下来，您似乎已经从 SearchThread 稍微改变了 NewSearchThread 的方法 - 因为 NewSearchThread 的构造函数采用 SearchQuery 参数，而 SearchThread 从 BlockingQueue 中获取 SearchQuery。

最后，您的 NewServerPool 类的方法与 ServerPool 不同，因为 ServerPool 的 run() 方法不断地将新的 SearchQuery 放入 BlockingQueue 中。相比之下，NewServerPool 的构造函数接受单个 SearchQuery 并且不对其执行任何操作。

像这样的东西如何帮助您开始:

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)));
        }
    }
}

请注意，我说的是“让您开始”...因为上面仍然需要一些补充，例如在停止处理请求时正确退出 run() 方法(但这就是另一个话题)。