Java怎么創(chuàng)建多線程服務(wù)器

本篇內(nèi)容主要講解“Java怎么創(chuàng)建多線程服務(wù)器”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“Java怎么創(chuàng)建多線程服務(wù)器”吧!

一個(gè)典型的單線程服務(wù)器示例如下：

while (true) {
    Socket socket = null;
    try {
        // 接收客戶連接
        socket = serverSocket.accept();
        // 從socket中獲得輸入流與輸出流，與客戶通信
        ...
    } catch(IOException e) {
        e.printStackTrace()
    } finally {
        try {
            if(socket != null) {
                // 斷開連接
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

服務(wù)端接收到一個(gè)客戶連接，就與客戶進(jìn)行通信，通信完畢后斷開連接，然后接收下一個(gè)客戶連接，假如同時(shí)有多個(gè)客戶連接請求這些客戶就必須排隊(duì)等候。如果長時(shí)間讓客戶等待，就會(huì)使網(wǎng)站失去信譽(yù)，從而降低訪問量。

一般用并發(fā)性能來衡量一個(gè)服務(wù)器同時(shí)響應(yīng)多個(gè)客戶的能力，一個(gè)具有好的并發(fā)性能的服務(wù)器，必須符合兩個(gè)條件：

• 能同時(shí)接收并處理多個(gè)客戶連接

• 對于每個(gè)客戶，都會(huì)迅速給予響應(yīng)

用多個(gè)線程來同時(shí)為多個(gè)客戶提供服務(wù)，這是提高服務(wù)器并發(fā)性能的最常用的手段，一般有三種方式：

• 為每個(gè)客戶分配一個(gè)工作線程

• 創(chuàng)建一個(gè)線程池，由其中的工作線程來為客戶服務(wù)

• 利用 Java 類庫中現(xiàn)成的線程池，由它的工作線程來為客戶服務(wù)

為每個(gè)客戶分配一個(gè)線程

服務(wù)器的主線程負(fù)責(zé)接收客戶的連接，每次接收到一個(gè)客戶連接，都會(huì)創(chuàng)建一個(gè)工作線程，由它負(fù)責(zé)與客戶的通信

public class EchoServer {
    private int port = 8000;
    private ServerSocket serverSocket;
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        System.out.println("服務(wù)器啟動(dòng)");
    }
    public void service() {
        while(true) {
            Socket socket = null;
            try {
                // 接教客戶連接
                socket = serverSocket.accept();
                // 創(chuàng)建一個(gè)工作線程
                Thread workThread = new Thread(new Handler(socket));
                // 啟動(dòng)工作線程
                workThread.start();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[])throws TOException {
        new EchoServer().service();
    }
    // 負(fù)責(zé)與單個(gè)客戶的通信   
    class Handler implements Runnable {
        private Socket socket;
        pub1ic Handler(Socket socket) {
            this.socket = socket;
        }
        private PrintWriter getWriter(Socket socket) throws IOException {...}
        private BufferedReader getReader(Socket socket) throws IOException {...}
        public String echo(String msg) {...}
        public void run() {
            try {
                System.out.println("New connection accepted" + socket.getInetAddress() + ":" + socket.getPort());
                BufferedReader br = getReader(socket);
                PrintWriter pw = getWriter(socket);
                String msg = null;
                // 接收和發(fā)送數(shù)據(jù)，直到通信結(jié)束
                while ((msg = br.readLine()) != null) {
                    System.out.println("from "+ socket.getInetAddress() + ":" + socket.getPort() + ">" + msg);
                    pw.println(echo(msg));
                    if (msg.equals("bye")) break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    // 斷開連接
                    if(socket != nulll) socket.close();
                } catch (IOException e) {
                    e,printStackTrace();
                }
            }
        }
    }
}

創(chuàng)建線程池

上一種實(shí)現(xiàn)方式有以下不足之處：

• 服務(wù)器創(chuàng)建和銷毀工作線程的開銷很大，如果服務(wù)器需要與許多客戶通信，并且與每個(gè)客戶的通信時(shí)間都很短，那么有可能服務(wù)器為客戶創(chuàng)建新線程的開銷比實(shí)際與客戶通信的開銷還要大

• 除了創(chuàng)建和銷毀線程的開銷，活動(dòng)的線程也消耗系統(tǒng)資源。每個(gè)線程都會(huì)占用一定的內(nèi)存，如果同時(shí)有大量客戶連接服務(wù)器，就必須創(chuàng)建大量工作線程，它們消耗了大量內(nèi)存，可能會(huì)導(dǎo)致系統(tǒng)的內(nèi)存空間不足

線程池中預(yù)先創(chuàng)建了一些工作線程，它們不斷地從工作隊(duì)列中取出任務(wù)，然后執(zhí)行該任務(wù)。當(dāng)工作線程執(zhí)行完一個(gè)任務(wù)，就會(huì)繼續(xù)執(zhí)行工作隊(duì)列中的下一個(gè)任務(wù)

線程池具有以下優(yōu)點(diǎn)：

• 減少了創(chuàng)建和銷毀線程的次數(shù)，每個(gè)工作線程都可以一直被重用，能執(zhí)行多個(gè)任務(wù)

• 可以根據(jù)系統(tǒng)的承載能力，方便調(diào)整線程池中線程的數(shù)目，防止因?yàn)橄倪^量系統(tǒng)資源而導(dǎo)致系統(tǒng)崩潰

public class ThreadPool extends ThreadGroup {
    // 線程池是否關(guān)閉
    private boolean isClosed = false;
    // 表示工作隊(duì)列
    private LinkedList<Runnable> workQueue;
    // 表示線程池ID
    private static int threadPoolID;
    // 表示工作線程ID
    // poolSize 指定線程池中的工作線程數(shù)目
    public ThreadPool(int poolSize) {
        super("ThreadPool-"+ (threadPoolID++));
        setDaemon(true);
        // 創(chuàng)建工作隊(duì)列
        workQueue = new LinkedList<Runnable>();
        for (int i = 0; i < poolSize; i++) {
            // 創(chuàng)建并啟動(dòng)工作線程
            new WorkThread().start(); 
        }
    }
    /**
     * 向工作隊(duì)列中加入一個(gè)新任務(wù)，由工作線程去執(zhí)行任務(wù)
     */
    public synchronized void execute(Runnable tank) {
        // 線程池被關(guān)則拋出IllegalStateException異常
        if(isClosed) {
            throw new IllegalStateException();
        }
        if(task != null) {
            workQueue.add(task);
            // 喚醒正在getTask()方法中等待任務(wù)的工作線限
            notify();
        }
    }
    /**
     * 從工作隊(duì)列中取出一個(gè)任務(wù)，工作線程會(huì)調(diào)用此方法
     */
    protected synchronized Runnable getTask() throws InterruptedException {
        while(workQueue,size() == 0) {
            if (isClosed) return null;
            wait(); // 如果工作隊(duì)列中沒有任務(wù)，就等待任務(wù)
        }
        return workQueue.removeFirst();
    }
    /**
     * 關(guān)閉線程池
     */
    public synchronized void close() {
        if(!isClosed) {
            isClosed = true;
            // 清空工作隊(duì)列
            workQueue.clear();
            // 中斷所有的工作線程，該方法繼承自ThreadGroup類
            interrupt();
        }
    }
    /**
     * 等待工作線程把所有任務(wù)執(zhí)行完
     */
    public void join() {
        synchronized (this) {
            isClosed = true;
            // 喚醒還在getTask()方法中等待任務(wù)的工作線程
            notifyAll();
        }
        Thread[] threads = new Thread[activeCount()];
        // enumerate()方法繼承自ThreadGroup類獲得線程組中當(dāng)前所有活著的工作線程
        int count = enumerate(threads);
        // 等待所有工作線程運(yùn)行結(jié)束
        for(int i = 0; i < count; i++) {
            try {
                // 等待工作線程運(yùn)行結(jié)束
                threads[i].join();
            } catch((InterruptedException ex) {}
        }
    }
    /**
     * 內(nèi)部類:工作線程
     */
    private class WorkThread extends Thread {
        public WorkThread() {
            // 加入當(dāng)前 ThreadPool 線程組
            super(ThreadPool.this, "WorkThread-" + (threadID++));
        }
        public void run() {
            // isInterrupted()方法承自Thread類，判斷線程是否被中斷
            while (!isInterrupted()) {
                Runnable task = null;
                try {
                    // 取出任務(wù)
                    task = getTask();
                } catch(InterruptedException ex) {}
                // 如果 getTask() 返回 nu11 或者線程執(zhí)行 getTask() 時(shí)被中斷，則結(jié)束此線程
                if(task != null) return;
                // 運(yùn)行任務(wù)，異常在catch代碼塊中被捕獲
                try {
                    task.run();
                } catch(Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }
}

使用線程池實(shí)現(xiàn)的服務(wù)器如下：

publlc class EchoServer {
    private int port = 8000;
    private ServerSocket serverSocket;
    private ThreadPool threadPool;	// 線程港
    private final int POOL_SIZE = 4;	// 單個(gè)CPU時(shí)線程池中工作線程的數(shù)目
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        // 創(chuàng)建線程池
        // Runtime 的 availableProcessors() 方法返回當(dāng)前系統(tǒng)的CPU的數(shù)目
        // 系統(tǒng)的CPU越多，線程池中工作線程的數(shù)目也越多
        threadPool= new ThreadPool(
        	Runtime.getRuntime().availableProcessors() * POOL_SIZE);
        System.out.println("服務(wù)器啟動(dòng)");
    }
    public void service() {
        while (true) {
            Socket socket = null;
            try {
                socket = serverSocket.accept();
                // 把與客戶通信的任務(wù)交給線程池
                threadPool.execute(new Handler(socket));
            } catch(IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[])throws TOException {
        new EchoServer().service();
    }
    // 負(fù)責(zé)與單個(gè)客戶的通信，與上例類似
    class Handler implements Runnable {...}
}

使用 Java 提供的線程池

java.util.concurrent 包提供了現(xiàn)成的線程池的實(shí)現(xiàn)，更加健壯，功能也更強(qiáng)大，更多關(guān)于線程池的介紹可以這篇文章

public class Echoserver {
    private int port = 8000;
    private ServerSocket serverSocket;
    // 線程池
    private ExecutorService executorService;
    // 單個(gè)CPU時(shí)線程池中工作線程的數(shù)目
    private final int POOL_SIZE = 4;
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        // 創(chuàng)建線程池
        // Runtime 的 availableProcessors() 方法返回當(dāng)前系統(tǒng)的CPU的數(shù)目
        // 系統(tǒng)的CPU越多，線程池中工作線程的數(shù)目也越多
        executorService = Executors.newFixedThreadPool(
        	Runtime.getRuntime().availableProcessors() * POOL_SIZE);
        System.out.println("服務(wù)器啟動(dòng)");
    }
    public void service() {
        while(true) {
            Socket socket = null;
            try {
                socket = serverSocket.accept();
                executorService.execute(new Handler(socket));
            } catch(IOException e) {
                e.printStackTrace();
            }
        }
    }
     public static void main(String args[])throws TOException {
        new EchoServer().service();
    }
    // 負(fù)責(zé)與單個(gè)客戶的通信，與上例類似
    class Handler implements Runnable {...}
}

使用線程池的注意事項(xiàng)

雖然線程池能大大提高服務(wù)器的并發(fā)性能，但使用它也存在一定風(fēng)險(xiǎn)，容易引發(fā)下面的問題：

• 死鎖

任何多線程應(yīng)用程序都有死鎖風(fēng)險(xiǎn)。造成死鎖的最簡單的情形是：線程 A 持有對象 X 的鎖，并且在等待對象 Y 的鎖，而線程 B 持有對象 Y 的鎖，并且在等待對象 X 的鎖，線程 A 與線程 B 都不釋放自己持有的鎖，并且等待對方的鎖，這就導(dǎo)致兩個(gè)線程永遠(yuǎn)等待下去，死鎖就這樣產(chǎn)生了

任何多線程程序都有死鎖的風(fēng)險(xiǎn)，但線程池還會(huì)導(dǎo)致另外一種死鎖：假定線程池中的所有工作線程都在執(zhí)行各自任務(wù)時(shí)被阻塞，它們都在等待某個(gè)任務(wù) A 的執(zhí)行結(jié)果。而任務(wù) A 依然在工作隊(duì)列中，由于沒有空閑線程，使得任務(wù) A 一直不能被執(zhí)行。這使得線程池中的所有工作線程都永遠(yuǎn)阻塞下去，死鎖就這樣產(chǎn)生了

• 系統(tǒng)資源不足

如果線程池中的線程數(shù)目非常多，這些線程就會(huì)消耗包括內(nèi)存和其他系統(tǒng)資源在內(nèi)的大量資源，從而嚴(yán)重影響系統(tǒng)性能

• 并發(fā)錯(cuò)誤

線程池的工作隊(duì)列依靠 wait() 和 notify() 方法來使工作線程及時(shí)取得任務(wù)，但這兩個(gè)方法都難以使用。如果編碼不正確，就可能會(huì)丟失通知，導(dǎo)致工作線程一直保持空閑狀態(tài)，無視工作隊(duì)列中需要處理的任務(wù)

• 線程泄漏

對于工作線程數(shù)目固定的線程池，如果工作線程在執(zhí)行任務(wù)時(shí)拋出 RuntimeException 或 Error，并且這些異?；蝈e(cuò)誤沒有被捕獲，那么這個(gè)工作線程就會(huì)異常終止，使得線程池永久地失去了一個(gè)工作線程。如果所有的工作線程都異常終止，線程池變?yōu)榭眨瑳]有任何可用的工作線程來處理任務(wù)

導(dǎo)致線程泄漏的另一種情形是，工作線程在執(zhí)行一個(gè)任務(wù)時(shí)被阻塞，比如等待用戶的輸入數(shù)據(jù)，但是由于用戶一直不輸入數(shù)據(jù)（可能是因?yàn)橛脩糇唛_了），導(dǎo)致這個(gè)工作線程一直被阻塞。這樣的工作線程名存實(shí)亡，它實(shí)際上不執(zhí)行任何任務(wù)了。假如線程池中所有的工作線程都處于這樣的阻塞狀態(tài)，那么線程池就無法處理新加入的任務(wù)了

• 任務(wù)過載

當(dāng)工作隊(duì)列中有大量排隊(duì)等候執(zhí)行的任務(wù)，這些任務(wù)本身可能會(huì)消耗太多的系統(tǒng)資源而引起系統(tǒng)資源缺乏

綜上所述，線程池可能會(huì)帶來種種風(fēng)險(xiǎn)，為了盡可能避免它們，使用線程池時(shí)需要遵循以下原則：

如果任務(wù) A 在執(zhí)行過程中需要同步等待任務(wù) B 的執(zhí)行結(jié)果，那么任務(wù) A 不適合加入線程池的工作隊(duì)列中。如集把像任務(wù) A 一樣的需要等待其他任務(wù)執(zhí)行結(jié)果的任務(wù)加入工作隊(duì)列中，就可能會(huì)導(dǎo)致線程池的死鎖

如果執(zhí)行某個(gè)任務(wù)時(shí)可能會(huì)阻塞，并且是長時(shí)間的阻塞，則應(yīng)該設(shè)定超時(shí)時(shí)間避免工作線程永久地阻塞下去而導(dǎo)致線程泄漏

了解任務(wù)的特點(diǎn)，分析任務(wù)是執(zhí)行經(jīng)常會(huì)阻塞的 IO 操作，還是執(zhí)行一直不會(huì)阻塞的運(yùn)算操作。前者時(shí)斷時(shí)續(xù)地占用 CPU，而后者對 CPU 具有更高的利用率。根據(jù)任務(wù)的特點(diǎn)，對任務(wù)進(jìn)行分類，然后把不同類型的任務(wù)分別加入不同線程池的工作隊(duì)列中，這樣可以根據(jù)任務(wù)的特點(diǎn)分別調(diào)整每個(gè)線程池

調(diào)整線程池的大小，線程池的最佳大小主要取決于系統(tǒng)的可用 CPU 的數(shù)目以及工作隊(duì)列中任務(wù)的特點(diǎn)。假如在一個(gè)具有 N 個(gè) CPU 的系統(tǒng)上只有一個(gè)工作隊(duì)列并且其中全部是運(yùn)算性質(zhì)的任務(wù)，那么當(dāng)線程池具有 N 或 N+1 個(gè)工作線程時(shí)，一般會(huì)獲得最大的 CPU 利用率

如果工作隊(duì)列中包含會(huì)執(zhí)行 IO 操作并經(jīng)常阻塞的任務(wù)，則要讓線程池的大小超過可用 CPU 的數(shù)目，因?yàn)椴⒉皇撬泄ぷ骶€程都一直在工作。選擇一個(gè)典型的任務(wù)，然后估計(jì)在執(zhí)行這個(gè)任務(wù)的過程中，等待時(shí)間（WT）與實(shí)際占用 CPU 進(jìn)行運(yùn)算的時(shí)間（ST）之間的比：WT/ST。對于一個(gè)具有 N 個(gè) CPU 的系統(tǒng)，需要設(shè)置大約 N(1+WT/ST) 個(gè)線程來保證 CPU 得到充分利用

避免任務(wù)過載，服務(wù)器應(yīng)根據(jù)系統(tǒng)的承受能力，限制客戶的并發(fā)連接的數(shù)目。當(dāng)客戶的并發(fā)連接的數(shù)目超過了限制值，服務(wù)器可以拒絕連接請求，并給予客戶友好提示。

到此，相信大家對“Java怎么創(chuàng)建多線程服務(wù)器”有了更深的了解，不妨來實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！