Öğeler bir süre döngüsüne eklendikçe sıradaki öğeler Öğe

Question

Bir süre döngüsünde UDP paketlerini dinlediğim bir yöntem var. Paketleri, geldikleri farklı bir sınıfta başka bir yöntem kullanarak ayrıştırmak ve uygulamanın her bir parçasında, her bir paketin farklı ayrıştırma ve analizlerini yapmak istiyorum. PacketParser yöntemlerinin, döngünün dışındaki Kuyruğu işlemesinin daha iyi olacağını düşünüyorum. Paketleri geldikçe Queue'a ekleyebilmemiz ve uygulamanın başka bir parçasının Kuyruğa geldikçe öğeleri dinleyebilmeleri ve döngüleri paket dinlemeye devam ederken orijinal olarak diğer eylemleri gerçekleştirmeleri mümkün olacaktır. kuyruk? Sırasıyla başka bir işleve sahip olmak ve paketleri işlemek istiyorum, Queue veya Stack'u izlemek için Java'da bir şey var mı? Bunu yapmanın daha iyi bir yolu var mı?

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

Tamam, bu nedenle üretici-tüketici deseni hakkında okuma bazı yaptı ve ne yaptığını bu yüzden burada bunu anladım. Temel olarak üretici-tüketici modeli üç şeyden oluşur: bir üretici, tüketici ve paylaşılan bir kuyruk. Bu bağlamda, PacketReader, ağ paketlerini alan ve bunları paylaşılan sıraya yerleştiren üreticidir. PacketParser, paylaşılan sıradaki paketleri işleyen tüketicidir. Bu nedenle, bir LinkedBlockingQueue örneğini oluşturdum ve paylaşılan sırayı tüketici (PacketReader) örneğine ve üreticinin bir örneğine (PacketParser) aktardım. Daha sonra tüketici ve üretici örnekleri her biri Thread sınıfının bir örneğine geçirilir. Son olarak, her iş parçacığı örneğinde start() yöntemini çağırın. Tam eğer put() yöntem kuyruğunu engeller ve boşsa take() kuyrukta eğer engeller çünkü

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 kullanmak nedenidir. Üretici ve tüketici sınıfları, Runnable arabirimini uygulamalı ve hiçbir parametre almayan run() adlı bir yöntem içermelidir.

Tüketici sınıfı

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()); 
      } 
     } 
    } 

Üretici sınıf

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()); 
       } 
      } 
     } 
    } 
}