Java套接字编程: server loop is getting executed infinitely

Question

虽然我明白这一点

while(true){
}

生成无限循环，我的理解是

while(true){
blockingCall()
}

由于阻塞调用的性质，即如果有 3 个调用，

允许此循环执行 x 次(x 可以介于 0 和达到给定机器资源限制的数字之间)对 blockingCall() 方法进行的调用从未返回，这意味着程序应该在那里等待。这是一个实现主题，它没有按照我期望的方式工作。我正在使用 Java 套接字实现客户端/服务器程序。 https://docs.oracle.com/javase/tutorial/networking/sockets/clientServer.html是了解我的客户端正在做什么的引用链接(它只是请求连接到在特定端口上运行的服务器并发送消息。服务器反转该消息并发送回客户端)。我正在尝试以某种方式实现服务器，以便限制该服务器允许的连接数。如果请求连接的客户端数量超过此限制，则其他请求将排队到最大限制。一旦超过这个最大限制，服务器就会简单地向日志写入一条消息，说明“不再接受任何连接”。下面是我的服务器程序:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.*;

public class MultithreadedServer {
    private static BlockingQueue<Socket> queuedSockets = new ArrayBlockingQueue<>(1);                  //max queued connections.
    private static Semaphore semaphoreForMaxConnectionsAllowed = new Semaphore(2);              //max active connections being served.

    private static void handleClientConnectionRequest(final Socket newSocketForClientConnection, final Semaphore maxConnectionSemaphore) {
        new Thread(new Runnable() {

            @Override
            public void run() {

                try (
                        BufferedReader socketReader = new BufferedReader(new InputStreamReader(newSocketForClientConnection.getInputStream()));
                        PrintWriter socketWriter = new PrintWriter(newSocketForClientConnection.getOutputStream(), true)
                ) {

                    maxConnectionSemaphore.acquire();

                    String serverMsg;
                    String clientMsg;

                    SocketAddress clientSocket = (InetSocketAddress) newSocketForClientConnection.getRemoteSocketAddress();

                    while ((clientMsg = socketReader.readLine()) != null) {
                        if (clientMsg.equalsIgnoreCase("quit")) {
                            maxConnectionSemaphore.release();
                            break;
                        }

                        System.out.println("client with socket " + clientSocket + " sent MSG : " + clientMsg);
                        serverMsg = reverseString(clientMsg);

                        socketWriter.println(serverMsg);
                    }

                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("Closing client upon client's request.");
                }
            }
        }).start();
    }

    private static String reverseString(String clientMsg) {
        synchronized (clientMsg) {
            StringBuffer stringBuffer = new StringBuffer();

            for (int i = clientMsg.length() - 1; i >= 0; i--) {
                stringBuffer.append(clientMsg.charAt(i));
            }

            return stringBuffer.toString();
        }
    }

    public static void main(String[] args) throws IOException {
        boolean shouldContinue = true;

        if (args.length != 1) {
            System.out.println("Incorrect number of arguments at command line");
            System.exit(1);
        }

        ServerSocket serverSocket = null;

        try {
            Integer portNumber = Integer.parseInt(args[0]);
            serverSocket = new ServerSocket(portNumber);
            int connectionNumber = 0;

            System.out.println("Server listening on port# : " + args[0]);

            //main thread...
            while (shouldContinue) {
                Socket newServerSocketForClientConnection = null;
                newServerSocketForClientConnection = queuedSockets.poll();

                if (newServerSocketForClientConnection == null) {
                    newServerSocketForClientConnection = serverSocket.accept();

                    connectionNumber++;
                    System.out.println("Created new socket upon client request. ConnectionCOunt = " + connectionNumber);

                    processConnection(newServerSocketForClientConnection);
                } else {
                    //i.e. queue has a socket request pending.
                    System.out.println("Picking queued socket..");
                    processConnection(newServerSocketForClientConnection);
                }
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (serverSocket != null) {
                serverSocket.close();
            }
        }
    }

    private static void processConnection(Socket newServerSocketForClientConnection) {

        if (semaphoreForMaxConnectionsAllowed.availablePermits() > 0) {
            handleClientConnectionRequest(newServerSocketForClientConnection, semaphoreForMaxConnectionsAllowed);
        } else {
            //System.out.println("Since exceeded max connection limit, adding in queue.");
            if (queuedSockets.offer(newServerSocketForClientConnection)) {
                System.out.println("connectionRequest queued because no more space on server. QueuedSocketList size : " + queuedSockets.size());
            }else{
                System.out.println("No space available for client connections. Can not be queued too.");
            }

        }

    }
}

当客户端请求的数量超过信号量限制时，通过此服务器观察到的输出(出于某种原因，我必须在我的程序中使用信号量并且不能将 ExecutorService 与 FixedThreadPool 一起使用):

我的问题是:看来 queuedSockets.poll() 似乎没有从 blockingQueue 中删除元素。这就是为什么我得到这个伪无限循环。任何线索为什么会这样？我检查了 blockingQueue 的文档，文档说 poll() 将“检索并删除该队列的头部”，但上面的程序似乎没有发生。

Answer 1

让我们逐步完成这个循环:

//main thread...
        while (shouldContinue) {
            Socket newServerSocketForClientConnection = null;
            // poll for a pending connection in the queue
            newServerSocketForClientConnection = queuedSockets.poll();

            // if a pending connection exists, go to else...
            if (newServerSocketForClientConnection == null) {
                ...
            } else {
                // queue has a socket request pending, so we process the request...
                System.out.println("Picking queued socket..");
                processConnection(newServerSocketForClientConnection);
            }
        }

然后在 processConnection() 中:

    // if there are no permits available, go to else...
    if (semaphoreForMaxConnectionsAllowed.availablePermits() > 0) {
        handleClientConnectionRequest(newServerSocketForClientConnection, semaphoreForMaxConnectionsAllowed);
    } else {
        // BlockingQueue.offer() puts this connection immediately back into the queue,
        // then the method exits
        if (queuedSockets.offer(newServerSocketForClientConnection)) {
            System.out.println("connectionRequest queued because no more space on server. QueuedSocketList size : " + queuedSockets.size());
        }else{
            System.out.println("No space available for client connections. Can not be queued too.");
        }

    }

之后，在循环的下一次迭代中:

//main thread...
        while (shouldContinue) {
            Socket newServerSocketForClientConnection = null;
            // poll immediately gets the same request that was 
            // removed in the previous iteration
            newServerSocketForClientConnection = queuedSockets.poll();

            // Once something is in the queue, this condition will
            // never be met, so no new incoming connections
            // can be accepted
            if (newServerSocketForClientConnection == null) {
                ...
            } else {
                // process the same request again, forever, or until
                // a connection is freed up. Meanwhile, all other
                // incoming requests are being ignored.
                System.out.println("Picking queued socket..");
                processConnection(newServerSocketForClientConnection);
            }
        }

所以并不是请求永远不会从队列中删除，它只是因为被信号量阻塞而在之后立即放回。