Java線程

Question

我寫了下面兩行

ServerSocket mobCom = new ServerSocket(9846); 
Socket server = mobCom.accept(); 

我希望創建一個新的TCP連接，並且連接到一個新的線程來處理。例如上面的代碼創建一個服務器套接字。而且有多個客戶。每當客戶端連接到服務器時，可能會創建一個新的線程來迎合來自該特定客戶端的請求。我如何實現相同的。

編輯

我也想一個線程池限制爲10個用戶。如果有更多用戶出現，我想向他們發送一條錯誤消息，而不處理來自他們的進一步請求。

Answer 1

你將不得不做下面的事情。 ServiceThread是線程將服務請求。

while (true) { 
       try { 
        Socket clientSocket = null; 
        if (null != serverSocket) { 
        clientSocket = serverSocket.accept(); 
        ServiceThread serverThread = new ServiceThread(clientSocket); // Create a new thread for each client 
        serverThread.start(); 
        } 
       } catch(Exception ex) { 
        System.out.println("Exception while accepting connection " + ex.getMessage()); 
        ex.printStackTrace(); 
       } 

Answer 2

您可以使用java util併發的SynchronousQueue實現預期的結果。 創建固定數量的工人。使用取調用啓動塊讀取到SynchronousQueue上。因此，如果所有工作人員都分別執行了一項工作並忙於處理它們（與套接字進行通信），則不會從SynchronousQueue讀取數據，因此同步隊列將失敗。檢查這個失敗（這意味着所有固定數量的工作人員都很忙，現在沒有鎖定到隊列中），拒絕下一個請求。

以下行代碼示例[未經測試 - 爲簡潔起見，請根據您的需要修改]。

public class BoundedServer 
{ 
    public static void main(String[] args) 
    { 
     /** 
     * Port to serve 
     */ 
     final int port = 2013; 

     /** 
     * Max Workers 
     */ 
     final int maxworkers = 10; 

     /** 
     * The server socket. 
     */ 
     ServerSocket mServerSocket = null; 

     /** 
     * Queue of work units to process if there is a worker available. 
     */ 
     final SynchronousQueue<WorkUnit> mQueueToProcess = new SynchronousQueue<WorkUnit>(); 

     /** 
     * Queue of work units to reject if there is no current worker available. 
     */ 
     final LinkedBlockingQueue<WorkUnit> mQueueToReject = new LinkedBlockingQueue<WorkUnit>(); 

     /** 
     * A thread pool to handle the work. 
     */ 
     final ExecutorService communicationservice = Executors.newFixedThreadPool(maxworkers); 

     /** 
     * Let a single thread take care of rejecting the requests when needed to do so. 
     */ 
     final ExecutorService rejectionservice = Executors.newSingleThreadExecutor(); 

     try 
     { 
      Runnable communicationlauncher = new Runnable() 
      { 
       public void run() 
       { 
        try 
        { 
         /** 
         * Set of workers to handle the work. 
         */ 
         final CommunicationWorker[] workers = new CommunicationWorker[maxworkers]; 

         communicationservice.invokeAll(Arrays.asList(workers)); 
        } 
        finally 
        { 
         communicationservice.shutdown(); 
        } 
       } 
      }; 

      new Thread(communicationlauncher).start(); 

      Runnable rejectionlauncher = new Runnable() 
      { 
       public void run() 
       { 
        try 
        { 
         RejectionWorker rejectionworker = new RejectionWorker(mQueueToReject); 

         rejectionservice.submit(rejectionworker); 
        } 
        finally 
        { 
         rejectionservice.shutdown(); 
        } 
       } 
      }; 
      new Thread(rejectionlauncher).start(); 

      mServerSocket = new ServerSocket(port); 

      while(true) 
      { 
       WorkUnit work = new WorkUnit(mServerSocket.accept()); 

       if(!mQueueToProcess.offer(work)) 
       { 
        mQueueToReject.add(work); 
       } 
      } 
     } 
     finally 
     { 
      try 
      { 
       mServerSocket.close(); 
      } 
     } 
    } 
} 


public class WorkUnit 
{ 
    private Socket mSocket = null; 

    public WorkUnit(Socket socket) 
    { 
     super(); 
     this.setSocket(socket); 
    } 

    public Socket getSocket() { 
     return mSocket; 
    } 

    public void setSocket(Socket mSocket) { 
     this.mSocket = mSocket; 
    } 
} 

public class CommunicationWorker 
implements Callable<Boolean> 
{ 
    private SynchronousQueue<WorkUnit> mQueueToProcess; 

    public CommunicationWorker(SynchronousQueue<WorkUnit> queueToProcess) 
    { 
     super(); 
     this.mQueueToProcess = queueToProcess; 
    } 

    @Override 
    public Boolean call() throws Exception 
    { 
     while(true) 
     { 
      WorkUnit work = mQueueToProcess.take(); 

      Socket socket = work.getSocket(); 

      // Code to handle socket communication and closure. 
      // Once the communication is finished, this thread will get blocked to mQueueToProcess. 
     } 
    } 
} 


public class RejectionWorker 
implements Callable<Boolean> 
{ 
    private LinkedBlockingQueue<WorkUnit> mQueueToReject; 

    public RejectionWorker(LinkedBlockingQueue<WorkUnit> queueToReject) 
    { 
     super(); 
     this.mQueueToReject = queueToReject; 
    } 

    @Override 
    public Boolean call() throws Exception 
    { 
     while(true) 
     { 
      WorkUnit work = mQueueToReject.take(); 

      Socket socket = work.getSocket(); 

      // Code to reject the request. 
     } 
    } 
}