Многопоточный UDP-сервер с использованием ThreadPool

Question

Привет, я создаю сеть Peer2Peer, используя сокет UDP, каждый раз, когда сервер получает сообщение от клиента, он создает поток для обработки запроса.am, пытающегося реализовать очередь ThreadPool для хранения запроса и ограничения количество потоков, выполняемых на машине, и всякий раз, когда поток свободен, мы получаем потоки из очереди (Producer/Consumer), я пробовал какое-то время, но новые потоки не добавляются в очередь, заканчивающуюся только в конце добавлен первый поток.

 
    public void run() {
 

BlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>(10); 

    ExecutorService tpes =Executors.newFixedThreadPool(10); 
    int server_port = 8767; 
    DatagramSocket s = null; 
    while(true){ 
     byte[] message = new byte[1024]; 
     DatagramPacket p = new DatagramPacket(message, message.length); 

     try{ 
      s = new DatagramSocket(server_port); 
     }catch (SocketException e) { 
      e.printStackTrace(); 
      System.out.println("Socket excep"); 
     } 

     try { 
      s.receive(p); 
      queue.put(new RequestHandler(p)); 
      tpes.execute(queue.take()); 

     }catch (Exception e) { 
      e.printStackTrace(); 
      System.out.println("IO EXcept"); 
     } 
     finally{ 
      s.close(); 
     } 
    } 
} 

Я не знаю, где пойдет не так? любая помощь была бы оценена

Answer 1

Вам не нужен Queue. ExecutorService имеет очередь для задач. Я также предлагаю не открывать новый сокет каждый раз, так как вы можете повторно использовать их.

Итоговый код может выглядеть примерно так.

public void run() { 
    ExecutorService tpes = Executors.newFixedThreadPool(10); 
    int server_port = 8767; 
    DatagramSocket s = null; 
    try { 
     s = new DatagramSocket(server_port); 
     while (true) { 
      byte[] message = new byte[1024]; 
      DatagramPacket p = new DatagramPacket(message, message.length); 
      try { 
       s.receive(p); 
       tpes.execute(new RequestHandler(p)); 
      } catch (IOException e) { 
       e.printStackTrace(); 
      } 
     } 
    } catch (SocketException e) { 
     e.printStackTrace(); 
     System.out.println("Socket excep"); 
    } finally { 
     if (s != null) s.close(); 
    } 
} 

RequestHandler будет работать на одной из 10 нитей, и если вы получите больше запросов, чем 10 потоков может обрабатывать вы будете иметь очередь ожидающих RequestHandler задач в рамках Executor.

Примечание: я не думаю, что имеет смысл запускать RequestHandler в пуле потоков. Обработка 1 кбайт данных невелика и, например, запись этих данных на жесткий диск происходит быстрее, чем вы можете получить пакеты. (если вы не на волокне, а ваш жесткий диск очень медленный).