我有一种方法可以在 while 循环中监听 UDP 数据包。我想在数据包到达时使用不同类中的另一种方法解析数据包，并在应用程序的另一部分对每个数据包进行许多不同的解析和分析。我认为让 PacketParser 方法在循环外处理 Queue 会更好。是否可以在数据包进入时将数据包添加到 Queue 中，然后让应用程序的另一部分在项目进入队列时监听项目并执行其他操作，因为原始 while 循环保持监听数据包并将它们添加到队列中？我想让另一个函数监视队列并处理数据包，Java 中是否有一些东西可以监视 Queue 或 Stack？有更好的方法吗？

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

最佳答案

好的，所以我阅读了一些有关生产者-消费者模式的资料并弄明白了，所以这就是我所做的。

基本上，生产者-消费者模式涉及三件事:生产者、消费者和共享队列。在此上下文中，PacketReader 是接收网络数据包并将其放入共享队列的生产者。 PacketParser 是处理共享队列中数据包的消费者。因此，我创建了一个 LinkedBlockingQueue 实例，并将该共享队列传递给一个消费者实例 (PacketReader) 和一个生产者实例 (PacketParser)。然后将消费者和生产者实例分别传递到 Thread 类的实例中。最后，在每个线程实例上调用 start() 方法。

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 的原因是 put() 方法会在队列满时阻塞，而 take() 会在队列满时阻塞如果为空则排队。生产者和消费者类需要实现 Runnable 接口(interface)并包含名为 run() 的不带参数的方法。

消费类

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

关于java - 在 while 循环中添加项目时处理队列中的项目，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/34795139/