TCP套接字服務器的CPU時間超過限制'

Question

這個問題提出了運行使用生產者/消費者設計創建了一個socket服務器的同時，程序日誌，錯誤cpu time limit exceeded墜毀。我還發現，cpu使用率超過90%的時間。這裏是服務器的代碼，它可能會出錯，我該如何優化？

我用這個方法queue以避免爲每個請求的那麼多的threads創建。

在主方法（主線程）

//holds socket instances 
ConcurrentLinkedQueue<Socket> queue = new ConcurrentLinkedQueue<>(); 

//create producer thread 
Thread producer = new Thread(new RequestProducer(queue)); 
//create consumer thread 
Thread consumer = new Thread(new RequestConsumer(queue)); 

producer.start(); 
consumer.start(); 

RequestProducer螺紋

//this holds queue instance coming from main thread 
ConcurrentLinkedQueue<Socket> queue 

//constructor, initiate queue 
public RequestProducer(
    ConcurrentLinkedQueue<Socket> queue 
) { 
    this.queue = queue; 
} 

public void run() { 
    try { 
     //create serversocket instance on port 19029 
     ServerSocket serverSocket = new ServerSocket(19029); 
     while (true) { 
      try { 
       //keep accept connections 
       Socket socket = serverSocket.accept(); 
       //add socket to queue 
       queue.offer(socket); 
      } catch (ConnectException ce) {//handle exception 
      } catch (SocketException e) {//handle exception 
      } 
     } 
    } catch (IOException ex) {//handle exception} 
} 

RequestConsumer螺紋

//this holds queue instance coming from main thread, same as requestproducer 
ConcurrentLinkedQueue<Socket> queue 

//constructor, initiate queue 
public RequestConsumer(
    ConcurrentLinkedQueue<Socket> queue 
) { 
    this.queue = queue; 
} 

public void run() { 
    try { 
     Socket socket = null; 
     while (true) { 
      //get head of the queue (socket instance) 
      socket = queue.poll(); 
      if (null != socket) { 
       //process data stream 
       String in = DataStreamUtil.parseAsciiSockStream(socket.getInputStream()); 
       //close socket conection 
       socket.close(); 
       //excecute database insert of processed data 
       excecuteDbInsert(in); 
      } 
     } 
    } catch (IOException | ParseException ex) {//handle exceptions} 
} 

的數據流分析器

public static String parseAsciiSockStream(InputStream in) throws IOException { 
    StringBuilder builder = new StringBuilder(); 
    if (null != in) { 
     byte[] b = new byte[BYTE_STREAM_MAX]; 
     int length = in.read(b); 
     for (int i = 0; i < length; i++) { 
      builder.append((char) (int) b[i]); 
     } 
     in.close(); 
    } 
    return builder.toString(); 
} 

Answer 1

CPU時間超過限制，由於積極的while(true)循環到你的消費者。下面是你如何解決問題的一個例子。

您可以在while循環添加簡單的Thread.sleep(1)進入消費或使用等待/通知模式來限制CPU的消耗。

RequestProducer線程

import java.io.IOException; 
import java.net.ConnectException; 
import java.net.ServerSocket; 
import java.net.Socket; 
import java.net.SocketException; 
import java.util.concurrent.ConcurrentLinkedQueue; 

public class RequestProducer implements Runnable { 
    //this holds queue instance coming from main thread 
    final ConcurrentLinkedQueue<Socket> queue; 

    //constructor, initiate queue 
    public RequestProducer(
      ConcurrentLinkedQueue<Socket> queue 
    ) { 
     this.queue = queue; 
    } 

    public void run() { 
     try { 
      //create serversocket instance on port 19029 
      ServerSocket serverSocket = new ServerSocket(19029); 
      while (true) { 
       try { 
        //keep accept connections 
        Socket socket = serverSocket.accept(); 
        //add socket to queue 
        queue.offer(socket); 
        synchronized (queue) { 
         System.out.println("notifying"); 
         queue.notify(); 
        } 
       } catch (ConnectException ce) {//handle exception 
       } catch (SocketException e) {//handle exception 
       } 
      } 
     } catch (IOException ex) {//handle exception} 
     } 

    } 
} 

RequestConsumer線程

import java.io.IOException; 
import java.net.Socket; 
import java.text.ParseException; 
import java.util.concurrent.ConcurrentLinkedQueue; 

public class RequestConsumer implements Runnable { 
    //this holds queue instance coming from main thread, same as requestproducer 
    final ConcurrentLinkedQueue<Socket> queue; 

    //constructor, initiate queue 
    public RequestConsumer(
      ConcurrentLinkedQueue<Socket> queue 
    ) { 
     this.queue = queue; 
    } 

    public void run() { 
     try { 
      Socket socket = null; 
      while (true) { 
       //get head of the queue (socket instance) 
       System.out.println("Waiting for new socket"); 
       synchronized (queue) { 
        queue.wait(); 
       } 
       System.out.println("Acquired new socket"); 

       socket = queue.poll(); 
       try { 
        if (null != socket) { 
         //process data stream 
         String in = DataStreamUtil.parseAsciiSockStream(socket.getInputStream()); 
         //close socket conection 
         socket.close(); 
         //excecute database insert of processed data 
         //excecuteDbInsert(in); 

         System.out.println(in); 
        } 
       } finally { 
        if (socket != null) { 
         socket.close(); 
        } 
       } 

      } 
     } catch (IOException ex) {//handle exceptions} 
     } catch (InterruptedException e) { 
      e.printStackTrace(); 
      Thread.currentThread().interrupt(); 
     } 
    } 

} 

的數據流分析器

import java.io.IOException; 
import java.io.InputStream; 

public class DataStreamUtil { 
    public static String parseAsciiSockStream(InputStream in) throws IOException { 
     StringBuilder builder = new StringBuilder(); 
     if (null != in) { 
      byte[] b = new byte[BYTE_STREAM_MAX]; 
      System.out.println("Waiting for input"); 
      int length = in.read(b); 
      System.out.println("Got input"); 
      for (int i = 0; i < length; i++) { 
       builder.append((char) (int) b[i]); 
      } 
      in.close(); 
     } 
     return builder.toString(); 
    } 
}