ホームページ > 記事 > Java > Java でマルチスレッドサーバーを作成する方法

Java でマルチスレッドサーバーを作成する方法

PHPz転載: 2023-05-10 15:58:141342ブラウズ

典型的なシングルスレッドサーバーの例は次のとおりです:

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

サーバーはクライアント接続を受信し、クライアントと通信し、通信の完了後に切断し、次のクライアント接続を受信します。同時にクライアント接続であるため、これらのクライアントは複数のクライアント接続要求をキューに入れる必要があります。顧客を長時間待たせると、Web サイトの信頼性が失われ、トラフィックが減少します。

同時実行パフォーマンスは、通常、同時に複数のクライアントに応答するサーバーの能力を測定するために使用されます。優れた同時実行パフォーマンスを持つサーバーは、次の 2 つの条件を満たしている必要があります。複数の顧客接続を同時に受信して処理します

顧客ごとにすぐに応答が返されます
複数のスレッドを使用して複数のサービスに対応します顧客は同時にサービスを提供します。これは、サーバーの同時実行パフォーマンスを向上させる最も一般的な手段です。通常、次の 3 つの方法があります:

各顧客にワーカースレッドを割り当てる

スレッドプールを作成すると、その作業スレッドが顧客にサービスを提供します
Java クラスライブラリの既存のスレッドプールを使用すると、その作業スレッドが顧客にサービスを提供します
各顧客にスレッドを割り当てる

public class EchoServer {
    private int port = 8000;
    private ServerSocket serverSocket;
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        System.out.println("服务器启动");
    }
    public void service() {
        while(true) {
            Socket socket = null;
            try {
                // 接教客户连接
                socket = serverSocket.accept();
                // 创建一个工作线程
                Thread workThread = new Thread(new Handler(socket));
                // 启动工作线程
                workThread.start();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[])throws TOException {
        new EchoServer().service();
    }
    // 负责与单个客户的通信   
    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;
                // 接收和发送数据，直到通信结束
                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();
                }
            }
        }
    }
}

スレッドプールの作成

以前の実装には次の欠点があります:

サーバーワーカースレッドの作成と破棄オーバーヘッドが大きいサーバーが多くのクライアントと通信する必要があり、各クライアントとの通信時間が非常に短い場合、クライアント用に新しいスレッドを作成するサーバーのオーバーヘッドが、実際にクライアントと通信するオーバーヘッド。

スレッドの作成と破棄のオーバーヘッドに加えて、アクティブなスレッドはシステムリソースも消費します。各スレッドは一定量のメモリを占有します。多数のクライアントが同時にサーバーに接続する場合、多数のワーカースレッドを作成する必要があります。ワーカースレッドは大量のメモリを消費し、システム内のメモリ領域が不足する可能性があります。 system.
一部のワーカースレッドはスレッドプール内に事前に作成されており、継続的にワークキューからタスクを取得して、そのタスクを実行します。ワーカースレッドはタスクの実行を終了すると、ワークキュー内の次のタスクの実行を継続します。

時間を短縮します。スレッドの作成と破棄に必要な時間は、各ワーカースレッドを常に再利用でき、複数のタスクを実行できます。

スレッドプール内のスレッドの数は、システムの状況に応じて簡単に調整できます。システムのリソースの過度の消費を防ぎ、システムのクラッシュを引き起こす可能性のある処理能力を備えています。

public class ThreadPool extends ThreadGroup {
    // 线程池是否关闭
    private boolean isClosed = false;
    // 表示工作队列
    private LinkedList<Runnable> workQueue;
    // 表示线程池ID
    private static int threadPoolID;
    // 表示工作线程ID
    // poolSize 指定线程池中的工作线程数目
    public ThreadPool(int poolSize) {
        super("ThreadPool-"+ (threadPoolID++));
        setDaemon(true);
        // 创建工作队列
        workQueue = new LinkedList<Runnable>();
        for (int i = 0; i < poolSize; i++) {
            // 创建并启动工作线程
            new WorkThread().start(); 
        }
    }
    /**
     * 向工作队列中加入一个新任务，由工作线程去执行任务
     */
    public synchronized void execute(Runnable tank) {
        // 线程池被关则抛出IllegalStateException异常
        if(isClosed) {
            throw new IllegalStateException();
        }
        if(task != null) {
            workQueue.add(task);
            // 唤醒正在getTask()方法中等待任务的工作线限
            notify();
        }
    }
    /**
     * 从工作队列中取出一个任务，工作线程会调用此方法
     */
    protected synchronized Runnable getTask() throws InterruptedException {
        while(workQueue,size() == 0) {
            if (isClosed) return null;
            wait(); // 如果工作队列中没有任务，就等待任务
        }
        return workQueue.removeFirst();
    }
    /**
     * 关闭线程池
     */
    public synchronized void close() {
        if(!isClosed) {
            isClosed = true;
            // 清空工作队列
            workQueue.clear();
            // 中断所有的工作线程，该方法继承自ThreadGroup类
            interrupt();
        }
    }
    /**
     * 等待工作线程把所有任务执行完
     */
    public void join() {
        synchronized (this) {
            isClosed = true;
            // 唤醒还在getTask()方法中等待任务的工作线程
            notifyAll();
        }
        Thread[] threads = new Thread[activeCount()];
        // enumerate()方法继承自ThreadGroup类获得线程组中当前所有活着的工作线程
        int count = enumerate(threads);
        // 等待所有工作线程运行结束
        for(int i = 0; i < count; i++) {
            try {
                // 等待工作线程运行结束
                threads[i].join();
            } catch((InterruptedException ex) {}
        }
    }
    /**
     * 内部类:工作线程
     */
    private class WorkThread extends Thread {
        public WorkThread() {
            // 加入当前 ThreadPool 线程组
            super(ThreadPool.this, "WorkThread-" + (threadID++));
        }
        public void run() {
            // isInterrupted()方法承自Thread类，判断线程是否被中断
            while (!isInterrupted()) {
                Runnable task = null;
                try {
                    // 取出任务
                    task = getTask();
                } catch(InterruptedException ex) {}
                // 如果 getTask() 返回 nu11 或者线程执行 getTask() 时被中断，则结束此线程
                if(task != null) return;
                // 运行任务，异常在catch代码块中被捕获
                try {
                    task.run();
                } catch(Throwable t) {
                    t.printStackTrace();
                }
            }
        }
    }
}

スレッドプールを使用して実装されたサーバーは次のとおりです。

publlc class EchoServer {
    private int port = 8000;
    private ServerSocket serverSocket;
    private ThreadPool threadPool;	// 线程港
    private final int POOL_SIZE = 4;	// 单个CPU时线程池中工作线程的数目
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        // 创建线程池
        // Runtime 的 availableProcessors() 方法返回当前系统的CPU的数目
        // 系统的CPU越多，线程池中工作线程的数目也越多
        threadPool= new ThreadPool(
        	Runtime.getRuntime().availableProcessors() * POOL_SIZE);
        System.out.println("服务器启动");
    }
    public void service() {
        while (true) {
            Socket socket = null;
            try {
                socket = serverSocket.accept();
                // 把与客户通信的任务交给线程池
                threadPool.execute(new Handler(socket));
            } catch(IOException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[])throws TOException {
        new EchoServer().service();
    }
    // 负责与单个客户的通信，与上例类似
    class Handler implements Runnable {...}
}

java. util.concurrent

パッケージは、より堅牢で強力な既製のスレッドプール実装を提供します。スレッドプールの詳細については、こちらを参照してください。記事

public class Echoserver {
    private int port = 8000;
    private ServerSocket serverSocket;
    // 线程池
    private ExecutorService executorService;
    // 单个CPU时线程池中工作线程的数目
    private final int POOL_SIZE = 4;
    public EchoServer() throws IOException {
        serverSocket = new ServerSocket(port);
        // 创建线程池
        // Runtime 的 availableProcessors() 方法返回当前系统的CPU的数目
        // 系统的CPU越多，线程池中工作线程的数目也越多
        executorService = Executors.newFixedThreadPool(
        	Runtime.getRuntime().availableProcessors() * POOL_SIZE);
        System.out.println("服务器启动");
    }
    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();
    }
    // 负责与单个客户的通信，与上例类似
    class Handler implements Runnable {...}
}

スレッドプールの使用に関する注意事項スレッドプールを使用するとサーバーの同時実行パフォーマンスが大幅に向上しますが、その使用には一定のリスクがあり、次の問題が発生しやすくなります。

#デッドロック

マルチスレッドアプリケーションにはデッドロックのリスクがあります。デッドロックを引き起こす最も単純な状況は、スレッド A がオブジェクト X のロックを保持し、オブジェクト Y のロックを待機しているのに対し、スレッド B はオブジェクト Y のロックを保持し、オブジェクト X のロックを待機していることです。スレッド A とスレッドB どちらも保持しているロックを解放せず、相手のロックを待つことになります。これにより、2 つのスレッドが永遠に待機することになり、デッドロックが発生します。
マルチスレッドプログラムにはデッドロックのリスクがありますが、スレッドプールまた、別の種類のデッドロックも発生します。スレッドプール内のすべてのワーカースレッドがそれぞれのタスクの実行中にブロックされていると仮定すると、それらはすべて特定のタスク A の実行結果を待っています。タスク A はまだワークキュー内にありますが、アイドル状態のスレッドがないため、タスク A を実行できません。これにより、スレッドプール内のすべてのワーカースレッドが永久にブロックされ、デッドロックが発生します。

システムリソースが不十分です

スレッドプールが存在する場合、システム内には非常に多くのスレッドがあり、これらのスレッドはメモリやその他のシステムリソースを含む大量のリソースを消費するため、システムのパフォーマンスに深刻な影響を及ぼします

同時実行エラー

スレッドプールのワークキューは、ワーカースレッドが時間内にタスクを取得できるようにするために、
wait()
メソッドと

メソッドに依存していますが、どちらのメソッドも実行が困難です。使用。正しくコーディングされていないと、通知が失われ、ワーカースレッドがアイドル状態のままになり、ワークキュー内で処理する必要があるタスクが無視される可能性があります

#スレッドリーク

#ワーカースレッドの数が固定されたスレッドプールの場合、タスクの実行時にワーカースレッドが RuntimeException または Error をスローし、これらの例外またはエラーがキャッチされない場合、ワーカースレッドは異常終了し、スレッドプールのエラーが発生します。ワーカースレッド。すべてのワーカースレッドが異常終了すると、スレッドプールが空になり、タスクを処理するための使用可能なワーカースレッドがなくなります。

スレッドリークにつながるもう 1 つの状況は、ユーザーがデータを入力するのを待機しているなど、タスクの実行中にワーカースレッドがブロックされているが、ユーザーがデータを入力していないため (おそらくユーザーが立ち去ったため)、これが発生することです。 work スレッドがブロックされ続けています。このようなワーカースレッドは名前だけが存在し、実際にはタスクを実行しません。スレッドプール内のすべてのワーカースレッドがそのようなブロック状態にある場合、スレッドプールは新しく追加されたタスクを処理できなくなります

#タスクの過負荷

要約すると、スレッドプールは次のようになります。さまざまなリスクを引き起こす可能性があります。それらをできるだけ回避するには、スレッドプールを使用するときに次の原則に従う必要があります。

タスク A が実行中にタスク B の実行結果を同期的に待つ必要がある場合の場合、タスク A はスレッドプールのワークキューに追加するのに適していません。タスク A のように、他のタスクの実行結果を待つ必要があるタスクをワークキューに追加すると、スレッドプールでデッドロックが発生する可能性があります。

タスクが実行されると、ブロックされる可能性があります。タスクがブロックされている場合は、ワーカースレッドが永続的にブロックされてスレッドリークが発生するのを防ぐために、タイムアウトを設定する必要があります。頻繁にブロックするか、決してブロックしない操作を実行します。前者は CPU を断続的に使用しますが、後者は CPU 使用率が高くなります。タスクの特性に従ってタスクを分類し、さまざまなタイプのタスクを異なるスレッドプールのワークキューに追加して、各スレッドプールをタスクの特性に応じて調整できるようにします。スレッドプールのサイズ。スレッドプールの最適なサイズは、主にシステム上で使用可能な CPU の数と、ワークキュー内のタスクの特性によって決まります。 N 個の CPU を備えたシステム上にワークキューが 1 つだけあり、それらがすべて計算タスクである場合、スレッドプールに N または N 1 個のワーカースレッドがあると、通常、最大 CPU 使用率が得られます

Ifワークキューには、IO 操作を実行して頻繁にブロックするタスクが含まれています。すべてのワーカースレッドが常に動作しているわけではないため、スレッドプールのサイズが使用可能な CPU の数を超えます。典型的なタスクを選択し、このタスクの実行中に待機時間 (WT) と CPU が計算に費やした実際の時間 (ST) の比率 (WT/ST) を推定します。 N 個の CPU を備えたシステムの場合、CPU が完全に利用されるようにするには、約 N(1 WT/ST) のスレッドをセットアップする必要があります。

タスクの過負荷を避けるために、サーバーは、顧客ベースの同時接続を制限する必要があります。システムの容量に依存します。クライアントの同時接続数が制限を超えると、サーバーは接続要求を拒否し、クライアントにわかりやすいプロンプトを表示することができます。

以上がJava でマルチスレッドサーバーを作成する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

声明：

この記事はyisu.comで複製されています。侵害がある場合は、admin@php.cn までご連絡ください。

前の記事：Java チャットウィンドウを作成するプロセスは何ですか?次の記事：Java チャットウィンドウを作成するプロセスは何ですか?

続きを見る

Java でマルチスレッドサーバーを作成する方法

スレッド プールの作成

パッケージは、より堅牢で強力な既製のスレッド プール実装を提供します。スレッド プールの詳細については、こちらを参照してください。記事

関連記事

スレッドプールの作成

パッケージは、より堅牢で強力な既製のスレッドプール実装を提供します。スレッドプールの詳細については、こちらを参照してください。記事