whileループ内でアイテムとしてのキュー項目を処理する

Question

私はwhileループでUDPパケットを待ち受ける方法があります。私は彼らが到着すると、別のクラスの別のメソッドを使用してパケットを解析し、アプリケーションの別の部分で各パケットの多くの異なる解析と分析を行います。私はPacketParserメソッドがループ外のQueueを処理する方が良いと考えています。 Queueにパケットを追加してから、アプリケーションの別の部分がキューに入り、元のように他のアクションを実行している間、パケットをリッスンして追加することができます待ち行列？私はキューを監視し、パケットを処理する別の機能を持っています。またはStackを監視するために、Javaに何かがありますか？これを行うより良い方法はありますか？

public void read(String multicastIpAddress, int multicastPortNumber) { 
     PacketParser parser = new PacketParser(logger); 
     InetAddress multicastAddress = null; 
     MulticastSocket multicastSocket = null; 
     final int PortNumber = multicastPortNumber; 
     try { 
      multicastAddress = InetAddress.getByName(multicastIpAddress); 
      multicastSocket = new MulticastSocket(PortNumber); 
      String hostname = InetAddress.getLocalHost().getHostName(); 
      byte[] buffer = new byte[8192]; 
      multicastSocket.joinGroup(multicastAddress); 
      System.out.println("Listening from " + hostname + " at " + multicastAddress.getHostName()); 
      int numberOfPackets = 0; 
      while (true) { 
       numberOfPackets++; 
       DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length); 
       multicastSocket.receive(datagramPacket); 
       // add to queue for another function to process the packets 

      } 
     } catch (SocketException socketException) { 
      System.out.println("Socket exception " + socketException); 
     } catch (IOException exception) { 
      System.out.println("Exception " + exception); 
     } finally { 
      if (multicastSocket != null) { 
       try { 
        multicastSocket.leaveGroup(multicastAddress); 
        multicastSocket.close(); 
       } catch (IOException exception) { 
        System.out.println(exception.toString()); 
       } 
      } 
     } 
    } 

Answer 1

私はプロデューサ - コンシューマパターンについていくつかの読書をしましたので、ここで私がしたことが分かりました。

基本的にプロデューサ - コンシューマパターンには、プロデューサ、コンシューマ、共有キューという3つの要素があります。このコンテキストでは、PacketReaderは、ネットワークパケットを取り込み、それらを共有キューに配置するプロデューサです。 PacketParserは、共有キュー内のパケットを処理するコンシューマです。そこで私はLinkedBlockingQueueのインスタンスを作成し、その共有キューをコンシューマ（PacketReader）のインスタンスとプロデューサ（PacketParser）のインスタンスに渡しました。その後、コンシューマインスタンスとプロデューサインスタンスは、それぞれThreadクラスのインスタンスに渡されます。最後に、各スレッドインスタンスでstart()メソッドを呼び出します。完全な場合put()方法は、キューをブロックしますし、空の場合take()がキュー場合はブロックしますので

public class Main { 

    public static void main(String[] args) { 
     BlockingQueue<Packet> queue = new LinkedBlockingQueue<>(); 
     ILogger logger = Injector.getLogger(); 

     Thread reader = new Thread(new PacketReader(logger, queue, "239.1.1.1", 49410)); 
     Thread parser = new Thread(new PacketParser(logger, queue)); 

     reader.start(); 
     parser.start(); 
    } 
} 

LinkedBlockingQueueを使用するための理由があります。プロデューサクラスとコンシューマクラスは、Runnableインターフェイスを実装し、パラメータを取らないrun()という名前のメソッドを含む必要があります。

Consumerクラス

public class PacketParser implements Runnable { 

    private ILogger logger; 
    private BlockingQueue<Packet> queue; 
    private boolean running = true; 

    public PacketParser(ILogger logger, BlockingQueue<Packet> queue) { 
     this.logger = logger; 
     this.queue = queue; 
    } 

    public void stop() { 
     running = false; 
    } 

    public void run() { 
     while (running) { 
      Packet packet; 
      try { 
       packet = queue.take(); 
       parse(packet); 
      } catch (InterruptedException exception) { 
       logger.Log(exception.getStackTrace().toString()); 
      } 
     } 
    } 

プロデューサークラス

public class PacketReader implements Runnable { 

    private ILogger logger; 
    private final Queue<Packet> queue; 
    private String multicastIpAddress; 
    private int multicastPortNumber; 
    private boolean running = true; 

    public PacketReader(ILogger logger, Queue<Packet> queue, String multicastIpAddress, int multicastPortNumber) { 
     this.logger = logger; 
     this.queue = queue; 
     this.multicastIpAddress = multicastIpAddress; 
     this.multicastPortNumber = multicastPortNumber; 
    } 

    public void stop() { 
     running = false; 
    } 

    public void run() { 
     InetAddress multicastAddress = null; 
     MulticastSocket multicastSocket = null; 
     try { 
      multicastAddress = InetAddress.getByName(multicastIpAddress); 
      multicastSocket = new MulticastSocket(multicastPortNumber); 
      String hostname = InetAddress.getLocalHost().getHostName(); 
      byte[] buffer = new byte[8192]; 
      multicastSocket.joinGroup(multicastAddress); 
      System.out.println("Listening from " + hostname + " at " + multicastAddress.getHostName()); 
      int numberOfPackets = 0; 

      while (running) { 
       numberOfPackets++; 
       DatagramPacket datagramPacket = new DatagramPacket(buffer, buffer.length); 
       multicastSocket.receive(datagramPacket); 
       Packet packet = new Packet(numberOfPackets, datagramPacket); 
       queue.add(packet); 
      } 
     } catch (SocketException socketException) { 
      System.out.println("Socket exception " + socketException); 
     } catch (IOException exception) { 
      System.out.println("Exception " + exception); 
     } finally { 
      if (multicastSocket != null) { 
       try { 
        multicastSocket.leaveGroup(multicastAddress); 
        multicastSocket.close(); 
       } catch (IOException exception) { 
        System.out.println(exception.toString()); 
       } 
      } 
     } 
    } 
}