java - 如何修复 'Server sending data faster than client can handle, server freezes'

Question

我在用 GameMaker Studio 制作的游戏中使用 java 服务器来促进在线多人游戏，玩家将数据发送到 java 服务器，java 服务器将处理数据并将其发送给玩家。问题是当互联网连接速度慢的玩家无法处理发送给它的数据量时，它会导致所有玩家的服务器卡住(服务器将不再处理其他玩家发送的任何数据).

我通过使用 NetLimiter 并将一台笔记本电脑的下载速度设置为 5 kb/s 来模拟较慢的互联网速度，同时在其他笔记本电脑上保持适当的速度。我已经尝试将 ACK 数据包从 Java 服务器发送到客户端，如果它在 1 秒内没有响应，则不会再向该客户端发送更多数据(最终客户端将被踢)。这减少了服务器卡住的机会，但它仍然会偶尔发生。

主.java

import java.net.Socket;
import java.net.SocketAddress;
import java.net.InetSocketAddress;
import java.io.IOException;
import java.util.HashMap;
import java.net.ServerSocket;
import java.net.SocketTimeoutException;

public class Main
{
    static ServerSocket serverSocket_;
    static HashMap<String, ServerInformation> servers_;
    static int verboseLevel_;
    static int threadTimeout_;
    static int masterPort_;
    static int serverNumber_;
    static int socketTimeOut_;

    static {
        Main.serverSocket_ = null;
        Main.servers_ = new HashMap<String, ServerInformation>();
        Main.verboseLevel_ = 5;
        Main.threadTimeout_ = 10;
        Main.masterPort_ = 6510;
        Main.serverNumber_ = 1;
        Main.socketTimeOut_ = 6000;
    }

    public static void main(final String[] args) {
        try {
            setupServerAndCleanup(Main.masterPort_);
            while (true) {
                handleIncomingConnection();
            }
        }
        catch (IOException e) {
            e.printStackTrace();
        }
    }

    static void setupServerAndCleanup(final int port) throws IOException {
        (Main.serverSocket_ = new ServerSocket()).setReuseAddress(true);
        Main.serverSocket_.bind(new InetSocketAddress(Main.masterPort_));
        System.out.println("Server socket up and running on port " + Main.masterPort_);
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                if (Main.serverSocket_ != null) {
                    try {
                        Main.serverSocket_.close();
                        System.out.println("Server socket closed, port released");
                    }
                    catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }));
    }

    static void handleIncomingConnection() throws IOException {
        final Socket clientSocket = Main.serverSocket_.accept();
        clientSocket.setSoTimeout(Main.socketTimeOut_);
        final ClientThread client = new ClientThread(clientSocket);
        client.start();
    }

}

客户端线程.java

案例 1 是处理向客户端发送数据的部分，特别是这一行:

thread2.out_.print(msg);

如果发送的数据多于一个客户端可以处理的数量，服务器也将卡住所有其他客户端。

    import java.util.Iterator;
    import java.io.IOException;
    import java.io.Reader;
    import java.io.InputStreamReader;
    import java.util.regex.Pattern;
    import java.io.BufferedReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    import java.net.SocketTimeoutException;

    public class ClientThread extends Thread
    {
    Socket clientSocket_;
    String clientIp_;
    String serverIp_;
    ServerInformation server_;
    PrintWriter out_;
    BufferedReader in_;
    boolean prepareTermination_;
    boolean terminated_;
    private static final Pattern numberPattern;

    static {
        numberPattern = Pattern.compile("\\d+");
    }

    public ClientThread(final Socket sock) {
        this.clientSocket_ = sock;
        this.clientIp_ = this.clientSocket_.getRemoteSocketAddress().toString();
        this.serverIp_ = null;
        this.server_ = null;
        this.prepareTermination_ = false;
        this.terminated_ = false;
    }

    @Override
    public void run() {
        try {
            this.out_ = new PrintWriter(this.clientSocket_.getOutputStream(), true);
            this.in_ = new BufferedReader(new InputStreamReader(this.clientSocket_.getInputStream()));

            long lastActionTime = System.currentTimeMillis();
            while (true) {

                if (this.in_.ready() || System.currentTimeMillis() - lastActionTime >= 1000 * Main.threadTimeout_) {
                    if (System.currentTimeMillis() - lastActionTime >= 1000 * Main.threadTimeout_) {
                        //this.logDebugMessage(3, "Thread was killed due to prolonged inactivity (" + Main.threadTimeout_ + " seconds)");
                        this.terminateThread();
                        return;
                    }

                    final String tempInputLine;
                    if(((tempInputLine = this.in_.readLine()) == null )){
                        this.terminateThread(); //end thread                        
                        return;             
                    }
                    else
                    {                   
                        lastActionTime = System.currentTimeMillis();                    
                        final String inputLine = tempInputLine.trim();
                        if (ClientThread.numberPattern.matcher(inputLine).matches()){
                        final int val = Integer.parseInt(inputLine);
                        switch (val) {
                          case 1: { //send data to other players
                                final int parseCount = Integer.parseInt(this.in_.readLine().trim());
                                final StringBuilder msg = new StringBuilder();
                                for (int j = 0; j < parseCount; ++j) {
                                    msg.append(String.valueOf(this.in_.readLine().trim()) + "|");
                                }
                                for (final ClientThread thread2 : this.server_.ipToClientThread_.values()) {
                                    if (thread2 != this) {
                                        thread2.out_.print(msg);
                                        thread2.out_.flush();
                                    }
                                }
                                //this.logDebugMessage(5, "Packet for others: '" + msg.toString() + "'");
                                break;
                            }   

                            case 2: { //remove game server
                                //this.logDebugMessage(1, "A game server has been deleted, ip: " + ipServer);
                                Main.servers_.remove(this.server_.ip_);
                                this.serverIp_ = null;
                                for (final ClientThread thread : this.server_.ipToClientThread_.values()) {
                                    thread.prepareTermination_ = true;
                                }
                                this.terminateThread();
                                return;
                            }
                            case 3: { //connect new client
                                final String ipServer = this.in_.readLine().trim();
                                final String ipClient = this.in_.readLine().trim(); 
                                this.logDebugMessage(1, "A client wishes to connect to a server, client: " + ipClient + ", server: " + ipServer);
                                final ServerInformation info = Main.servers_.getOrDefault(ipServer, null);
                                if (info == null) {
                                    System.out.println("Connection to the server failed, no such server in the server list");
                                   this.out_.print("*" + 1 + "|" + 1 + "~" + "|");
                                   this.out_.flush();                                   
                                break;
                                }
                                this.server_ = info;
                                this.server_.ipToClientThread_.put(ipClient, this);
                                this.logDebugMessage(1, "Connection success");
                                this.logDebugMessage(5,"Map: " + this.server_.ipToClientThread_);
                                    this.out_.print("*" + 1 + "|" + 2 + "~" + "|");
                                    this.out_.flush();
                                break;
                            }         
                            case 4: { //disconnect client
                                final String ipClient = this.in_.readLine().trim();
                                this.server_.ipToClientThread_.remove(ipClient);
                                this.logDebugMessage(1, String.valueOf(ipClient) + " disconnected from the server at " + this.server_.ip_);
                                this.serverIp_ = null;
                                this.terminateThread();
                                return;
                            }                   

                            case 5: { //host create new game
                                if (Main.serverNumber_ > 1000000) {
                                Main.serverNumber_ = 10;    
                                }
                                Main.serverNumber_ += 1;                                
                                final String ipServer = Integer.toString(Main.serverNumber_); //unique server number
                                final String ipHost =  this.in_.readLine().trim(); //host 
                                final String name = this.in_.readLine().trim(); //Server name
                                final String description = this.in_.readLine().trim(); //class
                                final String servervar1 = this.in_.readLine().trim(); //max players
                                final String servervar3 = this.in_.readLine().trim(); //current lap
                                final String servervar4 = this.in_.readLine().trim(); //total laps
                                final String servervar5 = this.in_.readLine().trim(); //status
                                final String servervar6 = this.in_.readLine().trim(); //Password
                                final String servervar7 = this.in_.readLine().trim(); //Online version
                                final String servervar8 = this.in_.readLine().trim(); //Game server
                                final long servervar9 = System.currentTimeMillis(); //server creation time
                                //this.logDebugMessage(1, "A game server has been registered, ip: " + ipServer + ", name: " + name + ", description: " + description + ", servervar1: " + servervar1);
                                final ServerInformation gameServer = new ServerInformation(name, servervar1, servervar3, servervar4, servervar5, servervar6, servervar7, servervar8, servervar9, ipHost, ipServer, this.clientSocket_, this.out_, this.in_);
                                gameServer.description_ = description;
                                gameServer.ipToClientThread_.put(ipHost, this);
                                this.server_ = gameServer;
                                Main.servers_.put(ipServer, gameServer);
                                this.serverIp_ = ipServer;
                                break;
                            }                               
                            default: {
                                this.logDebugMessage(0, "Unrecognized case: '" + inputLine + "', " + val);
                                break;
                            }
                        }
                    }
                    else if (inputLine.length() > 0) {
                        this.logDebugMessage(0, "Unformated '" + inputLine + "'");
                        if (this.server_ != null) {
                            this.server_.out_.print(inputLine);
                            this.server_.out_.flush();
                        }
                    }
                    if (this.prepareTermination_) {
                        this.terminateThread();
                        return;
                    }
                    continue;
                    }
                }
            }
        }
        catch (SocketTimeoutException e) {
            e.printStackTrace();
            try {
                this.terminateThread();
            }
            catch (IOException e2) {
                e2.printStackTrace();
            }
        }
        catch (IOException e3) {
            e3.printStackTrace();
            try {
                this.terminateThread();
            }
            catch (IOException e4) {
                e4.printStackTrace();
            }
        }
    }

    //debug messages
    void logDebugMessage(final int requiredVerbose, final String msg) {
        if (Main.verboseLevel_ >= requiredVerbose) {
            System.out.println("[" + this.clientIp_ + "]  " + msg);
        }
    }

    //terminate thread
    void terminateThread() throws IOException {
        if (!this.terminated_) {
            if (this.serverIp_ != null) {
                Main.servers_.remove(this.serverIp_);
            }
            this.clientSocket_.close();
            this.in_.close();
            this.out_.close();
            this.logDebugMessage(3, "Cleanup successful");
            this.terminated_ = true;
        }
    }
}

如果发送给客户端的数据超过其处理能力，如何避免服务器卡住，以便服务器可以继续向其他客户端发送数据？

编辑 所以我尝试使用 ExecutorService，但我一定是做错了什么，因为 java 服务器没有发送任何数据。

for (final ClientThread thread2 : this.server_.ipToClientThread_.values()) {
    if (thread2 != this) {
        executorService = Executors.newSingleThreadExecutor();

        executorService.execute(new Runnable() {
            public void run() {
                thread2.out_.print(msg);
                thread2.out_.flush();
            }
        });

        executorService.shutdown();     
    }
}

如果你能告诉我如何以正确的方式实现 ExecutorService 就太好了。

Answer 1

如果客户端处理的延迟无关紧要，这部分应该在每个客户端的不同流程执行中完成:

for (final ClientThread thread2 : this.server_.ipToClientThread_.values()) { 
        thread2.out_.print(msg);
        thread2.out_.flush();            
}

例如:

for (final ClientThread thread2 : this.server_.ipToClientThread_.values()) { 
      if (thread2 != this) {
            new Thread(()-> {   
                                thread2.out_.print(msg);
                                thread2.out_.flush();
                            })
                      .start(); 
        }
}

请注意，创建线程是有成本的。使用 ExecutorService 可能是一个更好的主意。