多线程UDP服务器Java

Question

我有两个线程，一个在一个套接字上监听并添加到队列中，另一个从队列中减去并提交处理。第二个线程在队列为空时Hibernate。这个睡眠不知怎么会影响第一个线程，也就是说，如果您移除睡眠或使它变大，那么第一个线程的socket.receive中的延迟就会增加。如果我保持尽可能低的睡眠，它会变得更好，但不是完美的。我做错了什么？

private DatagramSocket socket;
private boolean running, listenthreadwork;
private byte[] buf = new byte[256];
private List<IUDPListener> listeners = new ArrayList<IUDPListener>();
private Thread runnerThread, listenerThread;
private ConcurrentLinkedQueue<MyObject> list = new ConcurrentLinkedQueue<MyObject>();

public void init(int port) throws SocketException
{
    socket = new DatagramSocket(port);
    
    runnerThread = new Thread(this::listenLoopUDP);
    runnerThread.setName("listenLoopUDP");
    
    listenerThread = new Thread(this::listenerThreadUDP);
    listenerThread.setName("listenerThreadUDP");
    
    running = true;
    listenthreadwork = true;
    runnerThread.start();
    listenerThread.start();
}

private void listenerThreadUDP() {
    while (listenthreadwork == true) {
        MyObjectinfo = null;
        synchronized (list) {
            if (!list.isEmpty()) {
                info = list.poll();
            }
        }
        if (info != null) {
            for (IUDPListener listener : listeners) {
                listener.msgReceived(info);
            }
        } else {
            try {
                Thread.sleep(1);//Somehow affects listenLoopUDP
            } catch (InterruptedException e) {
                Log.write(e);
            }
        }
    }
}

public void listenLoopUDP() {
    while (running) {
        DatagramPacket packet = new DatagramPacket(buf, buf.length);

        try {
            socket.receive(packet);
        } catch (IOException e) {
            if (socket.isClosed()) {
                running = false;
                continue;
            } else {
                Log.write(e);
                socket.close();
            }
        }
        
        String received = new String(packet.getData());

        MyObject info = new MyObject(received);
        synchronized (list) {
            list.offer(info);
        }
    }

    listenthreadwork = false;
    try {
        listenerThread.join();
    } catch (InterruptedException e) {
        Log.write(e);
    }
}

Answer 1

实际上，您不需要Hibernate，而是使用适当的队列类，如LinkedBlockingQueue，我还删除了这些标志，因为您也不需要它们，使用interrupt()来停止一个阻塞的线程，等待队列元素：

private DatagramSocket socket;
private byte[] buf = new byte[256];
private List<IUDPListener> listeners = new ArrayList<IUDPListener>();
private Thread runnerThread, listenerThread;
private LinkedBlockingQueue<MyObject> list = new LinkedBlockingQueue<MyObject>();

public void init(int port) throws SocketException
{
    socket = new DatagramSocket(port);
    
    runnerThread = new Thread(this::listenLoopUDP);
    runnerThread.setName("listenLoopUDP");
    
    listenerThread = new Thread(this::listenerThreadUDP);
    listenerThread.setName("listenerThreadUDP");
    
    runnerThread.start();
    listenerThread.start();
}

private void listenerThreadUDP() {
    try {
        while (true) {
            MyObject info=list.take();
            for (IUDPListener listener : listeners) {
                listener.msgReceived(info);
            }
        }
    } catch (InterruptedException ex) {
        //Just quit
    }
}

public void listenLoopUDP() {
    try {
        while (true) {
            DatagramPacket packet = new DatagramPacket(buf, buf.length);
            socket.receive(packet);
            String received = new String(packet.getData());
            MyObject info = new MyObject(received);
            list.put(info);
        }
    } catch (IOException e) {
        Log.write(e);
    } catch (InterruptedException e) {
        Log.write(e);
    } finally {
        //Any exception above (or a runtime one) will activate this block where we do the cleanup and interrupt the other running thread
        listenerThread.interrupt();
        socket.close();
    }
}