使用线程池处理客户端的回显服务器答案

【问题标题】：a echo server which handle clients with pool of thread使用线程池处理客户端的回显服务器
【发布时间】：2019-07-20 06:24:22
【问题描述】：

客户端连接有效（我使用 telnet），但是当我与任何客户端编写消息时，什么都没有发生 - 即使是空字符的情况（用于断开连接）。我不明白为什么。我通过 args[] 获得容量和端口，然后启动服务器。

我已经测试了一个更简单的版本，只有一个可以处理一个客户端的服务器，它可以工作。

public class EchoClient extends Thread {
    EchoServerForPool serv;
    BufferedReader inchan;
    DataOutputStream outchan;
    Socket socket;
    int port;

    public EchoClient(EchoServerForPool serv) {
        // TODO Auto-generated constructor stub
        this.serv = serv;
    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        Socket s;
        while (true) {
            synchronized (this.serv) {
                if (this.serv.stillWaiting() == 0) {
                    try {
                        this.serv.wait();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }

                s = this.serv.removeFirstSocket();
                serv.newConnect();
            }

            try {
                inchan = new BufferedReader(new InputStreamReader(s.getInputStream()));
                outchan = new DataOutputStream(s.getOutputStream());
                String message = inchan.readLine();
                if (message.equals("")) {
                    System.out.println("fin de connection");
                    break;
                }

                outchan.writeChars(message + "\n");
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } finally {
                try {
                    s.close();
                    synchronized (serv) {
                        serv.clientLeft();
                    }
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

        }
    }

    public class EchoServerForPool extends Thread {
        ArrayList<EchoClient> clients;
        ArrayList<Socket> sockets;
        Socket client;
        int nbLocalhost = 0;
        int capacity, port, nbConnectedClient, nbWaitingSocket;

        public EchoServerForPool(int capacity, int port) {
            // TODO Auto-generated constructor stub
            this.capacity = capacity;
            this.port = port;
            clients = new ArrayList<EchoClient>(capacity);
            sockets = new ArrayList<Socket>();
            for (int i = 0; i < clients.size(); i++) {
                EchoClient ec_i = new EchoClient(this);
                clients.add(ec_i);
                ec_i.start();
            }
        }

        public Socket removeFirstSocket() {
            Socket res = sockets.get(0);
            sockets.remove(0);
            return res;
        }

        public void newConnect() {
            nbConnectedClient++;
            nbWaitingSocket--;
        }

        public int stillWaiting() {
            return nbWaitingSocket;
        }

        public void clientLeft() {
            nbConnectedClient--;
        }

        @Override
        public void run() {
            // TODO Auto-generated method stub
            try {
                ServerSocket serv = new ServerSocket(this.port);
                while (true) {
                    this.client = serv.accept();
                    synchronized (this) {
                        nbLocalhost++;
                        System.out.println(client.getInetAddress().getHostName() + "-" + nbLocalhost + " connected");
                        sockets.add(client);
                        nbWaitingSocket++;
                        notify();
                    }
                }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
}

public class EchoPoolThread {
    public static void main(String[] args) {
        int port = Integer.parseInt(args[0]);
        int capacity = Integer.parseInt(args[1]);
        EchoClient.EchoServerForPool serveur = new EchoClient.EchoServerForPool(capacity, port);
        System.out.println("start server");
        serveur.start();
    }
}

编辑：问题是我用 0 而不是容量的 clients.size() 进行迭代，以填充我的客户列表....

【问题讨论】：

我认为您应该在打开新套接字时初始化 EchoClient 类，而不是在启动时。另外，您确定需要所有这些同步块吗？您可能正在处理死锁。
@devgianlu 你有一个我可以有死锁的例子吗？我绝对需要这两个同步块，因为所有客户端都可以访问服务器属性，我必须保护它们。
我无法为你调试代码（太费时间），你通常可以通过使用断点或简单的println语句来找到死锁。考虑从我在答案中提供给您的代码重新开始，因为您的代码非常复杂。
@devgianlu 首先，如果您无法给出导致死锁的执行示例，您不能说“我无法为您调试代码”。因为我没有看到它我花了很多时间在这些程序上。其次，在您的代码中，您创建的线程与客户端连接的线程一样多，如果它们中的任何一个都没有与服务器断开连接，并且您添加了一个新客户端，那么您将 arraylist 扩展到我决定声明的容量之外。所以基本上，你创建的线程和客户端一样多，或者我不明白，但这似乎不太好
@devgianlu 我说这不好，因为这里的目标是避免创建与客户端连接的线程一样多的线程，它非常无效，这就是我创建线程池的原因..

标签： java multithreading server synchronized

【解决方案1】：

我已经稍微重构了您的代码以获得有效的解决方案。您可以从那里创建一个池，现在客户端的数量不受限制。

import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Gianlu
 */
public class Main {
    public static void main(String[] args) {
        int port = Integer.parseInt(args[0]);
        int capacity = Integer.parseInt(args[1]);
        EchoServerForPool serveur = new EchoServerForPool(capacity, port);
        System.out.println("start server");
        serveur.start();
    }

    public static class EchoServerForPool extends Thread {
        private final int port;
        private final ExecutorService executor;
        private final ArrayList<EchoClient> clients;
        private int nbLocalhost = 0;

        public EchoServerForPool(int capacity, int port) {
            this.port = port;

            clients = new ArrayList<>();
            executor = Executors.newFixedThreadPool(capacity);
        }

        @Override
        public void run() {
            try {
                ServerSocket serv = new ServerSocket(this.port);

                while (true) {
                    Socket socket = serv.accept();
                    EchoClient client = new EchoClient(this, socket);
                    executor.submit(client);
                    clients.add(client);
                    nbLocalhost++;
                    System.out.println(socket.getInetAddress().getHostName() + "-" + nbLocalhost + " connected");
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        public void terminated(EchoClient client) {
            clients.remove(client);
        }
    }

    public static class EchoClient implements Runnable {
        private final EchoServerForPool pool;
        private final Socket socket;
        private final BufferedReader inchan;
        private final DataOutputStream outchan;

        public EchoClient(EchoServerForPool pool, Socket socket) throws IOException {
            this.pool = pool;
            this.socket = socket;
            this.inchan = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            this.outchan = new DataOutputStream(socket.getOutputStream());
        }

        @Override
        public void run() {
            try {
                while (true) {
                    String message = inchan.readLine();
                    if (message.equals("")) {
                        System.out.println("fin de connection");
                        break;
                    }

                    outchan.writeChars(message + "\n");
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            pool.terminated(this);
        }
    }
}

【讨论】：