コンシューマ/プロデューサデザインを使用するTCPソケットサーバー

Question

コンシューマ/プロデューサデザインを使用して作成したソケットサーバーを実行しているときに、この問題が発生し、エラーがログでエラーcpu time limit exceededでクラッシュしました。私もcpuの使用が90%以上であることがわかりました。ここではサーバーのコードは、何が間違って行くことができ、どのように私はこれを最適化することができますか？

私はこのリクエストを毎回threadsという非常に多く作成することを避けるために、このqueueというアプローチを使用しました。 mainメソッドで

（メインスレッド）

//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)をコンシューマに追加するか、wait/notifyパターンを使用して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(); 
    } 
}