Daemon Wątki, liczba wątków i łączna liczba rozpoczętych wątków

Question

Mam prosty kod dla wielowątkowego serwera echa w Javie (zwraca to, co otrzymałem z powrotem do klientów). Zajmuję się profilowaniem różnych zasobów serwera, w tym statystykami wątków. Poniżej przedstawiamy niektóre z tych statystyk według liczby podłączonych klientów. Moje pytania dotyczą linii bazowej (liczba klientów 0) w porównaniu z liniami nie-bazowymi!

1) dlaczego, gdy pojedynczy klient łączy się, całkowita liczba wątków wzrasta o 2? (dla reszty sensowny jest wzrost o 1)

2) Jakie są dwa wątki nie-demonów ?! I dlaczego demona początkowo inkrementuje o 1, a następnie jest naprawiony?

Czy są one losowe ?!

# clients     0 1 2 3 4 5 6 7 8 9 10 

Total Started Thread Count 15 18 19 20 21 22 23 24 25 26 27 
Thread count    14 16 17 18 19 20 21 22 23 24 25 
Peak thread count   14 16 17 18 19 20 21 22 23 24 25 
Daemon thread count   12 13 13 13 13 13 13 13 13 13 13 

Oto fragment kodu serwera. Używam zarówno RMI (dla klientów do odpytywania wiadomości) i Server Socket (dla klientów do wysyłania wiadomości). Jeśli potrzebujesz innych zajęć, daj mi znać.

package test; 

import java.io.BufferedReader; 
import java.io.IOException; 
import java.io.InputStreamReader; 
import java.io.PrintWriter; 
import java.net.ServerSocket; 
import java.net.Socket; 
import java.rmi.registry.LocateRegistry; 
import java.rmi.registry.Registry; 
import java.rmi.server.UnicastRemoteObject; 
import java.util.Vector; 

public class ServerRMI extends Thread implements Hello { 
    //centralized token manager runs polling server and socket server to receive updated tokens 
    static Vector<String> tokenList= new Vector<String>(); 
    protected Socket clientSocket; 
    static int RMIRegistryPort=9001; 
    static int SocketServerPort=9010; 

    public static void main(String[] args) throws IOException { 
     try { 
      ServerRMI obj = new ServerRMI(); 
      Hello stub = (Hello) UnicastRemoteObject.exportObject(obj, 0); 

      // Bind the remote object's stub in the registry 
      Registry registry = LocateRegistry.createRegistry(RMIRegistryPort); 
      registry.bind("Hello", stub); 
      System.err.println("Server ready"); 
     } catch (Exception e) { 
      System.err.println("Server exception: " + e.toString()); 
      e.printStackTrace(); 
     } 

     ServerSocket serverSocket = null; 
     //initialize token list 
     //A needs to execute first 
     tokenList.add(0,"0"); 

     try { 
      serverSocket = new ServerSocket(SocketServerPort); 
      System.out.println("Connection Socket Created"); 
      try { 
       while (true) { 
        System.out.println("Waiting for Connection"); 
        new ServerRMI(serverSocket.accept()); 
       } 
      } catch (IOException e) { 
       System.err.println("Accept failed."); 
      } 
     } catch (IOException e) { 
      System.err.println("Could not listen on port: "+SocketServerPort); 
     } finally { 
      try { 
       serverSocket.close(); 
      } catch (IOException e) { 
       System.err.println("Could not close port: "+SocketServerPort); 
      } 
     } 
    } 

    private ServerRMI(Socket clientSoc) { 
     clientSocket = clientSoc; 
     start(); 
    } 

    public ServerRMI() {}{ 
     // TODO Auto-generated constructor stub 
    } 

    public void run() { 
     System.out.println("New Communication Thread Started"); 

     try { 
      PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), 
        true); 
      BufferedReader in = new BufferedReader(new InputStreamReader(
        clientSocket.getInputStream())); 

      String inputLine; 

      while ((inputLine = in.readLine()) != null) { 
       tokenList.add(0,inputLine); 
       System.out.println("Server received: " + inputLine); 
//     System.out.println(" ququ size: "+queue.size()); 
       out.println(inputLine); 

       if (inputLine.equals("Bye.")) 
        break; 
      } 

      out.close(); 
      in.close(); 
      clientSocket.close(); 
     } catch (IOException e) { 
      System.err.println("Problem with Communication Server"); 
     } 
    } 

    public String pollServer() { 
     if(!tokenList.isEmpty()){ 
      String data = tokenList.get(0); 
      System.out.println("Poll data: "+data); 
      return data; 
     } else{ 
      return tokenList.size()+""; 
     } 
    } 
} 

Answer 1

Nie jestem pewien, czego używasz do otrzymywania tych połączeń, ale zazwyczaj ram, które obsługują połączenia TCP non-blocking jak Netty użyć wątek główny słuchać portu i puli wątków do obsługi połączeń przychodzących. Oznacza to, że przynajmniej otworzy 2 wątki dla przychodzącego połączenia, jeśli pula wątków jest ograniczona do 1 wątku.

Zobacz tę stronę example ze strony netty, na której są używane 2 NioEventLoopGroups podczas ładowania serwera.

2 wątków są niezbędne, aby nie blokować ruchu przychodzącego.

Answer 2

Mam nadzieję, że przejrzałeś nazwy wątków w programie profilującym VisualVM. Zazwyczaj nazwy wątków dają wyobrażenie o tym, co się zaczęło.

ad 1) może to być jakiś wątek TCP/IP (czystsze, poller), który zostanie utworzony, gdy tylko uzyskasz połączenie TCP/IP z zewnątrz.

Co się dzieje, gdy klienci wracają do 0. Czy dodatkowy wątek znika?

Odpowiedzi mogą bardzo różnić się w zależności od dostawcy i wersji JVM. Więc myślę, że musisz zajrzeć do wnętrza JVM o gniazdach i wątkach.

Przykład poniżej jest bardziej prosto do przodu (bez RMI)

import java.net.*; // for Socket, ServerSocket, and InetAddress 
import java.io.*; // for IOException and Input/0utputStream 

public class TCPEchoServer { 

    private static final int BUFSIZE = 32; // Size of receive buffer 
    public static void main(String[] args) throws lOException { 
     if (args.length != i) // Test for correct # of args 
      throw new lllegalArgumentException("Parameter(s): <Port>"); 
     int servPort = Integer.parselnt(args[0]); 
     // Create a server socket to accept client connection requests 
     ServerSocket servSock = new ServerSocket(servPort); 
     int recvMsgSize; // Size of received message 
     byte[] byteBuffer = new byte[BUFSlZE]; // Receive buffer 

    } 

    for (;;) { // Run forever, accepting and servicing connections 
     Socket clntSock = servSock.accept(); // Get client connection 
     System.out.println("Handling client at " + 
      clntSock.getInetAddress().getHostAddress() + " on port " + 
      clntSock, getPort()); 
     InputStream in = clntSock, getlnputStream(); 
     OutputStream out = clntSock.getOutputStream(); 
     // Receive until client closes connection, indicated by-i return 
     while ((recvMsgSize = in .read(byteBuffer)) != -I) 
      out.write(byteBuffer, O, recvMsgSize); 
     clntSock, close(); 
    } 
}