java - 停止 DatagramSocket 线程

Question

我目前正在开发一个 java 网络库，并且遇到了 DatagramSocket 的问题。我有一个线程不断监听 UDP 请求，每当我停止它并关闭其关联的 DatagramSocket 时，都会有一个线程无论如何都不会停止。

这是重现该问题的代码示例:

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class UDPListener implements Runnable {

    private volatile boolean isRunning;
    private volatile boolean clientClosed;
    private int port;
    private DatagramSocket client;

    public UDPListener(int listeningPort) {
        this.port = listeningPort;
    }

    public void open() throws SocketException {
        this.clientClosed = false;
        if (this.client != null) {
            this.client.close();
        }
        this.client = new DatagramSocket(this.port);
    }

    @Override
    public void run() {
        this.isRunning = true;
        this.clientClosed = false;
        DatagramPacket packet = new DatagramPacket(new byte[256], 256);
        while (this.isRunning) {

            try {
                this.client.receive(packet);
            } catch (SocketException e) {
                if (!this.clientClosed) {
                    e.printStackTrace();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        this.client.disconnect();
        System.out.println("stopped");
    }

    public void stop() {
        this.isRunning = false;
        this.clientClosed = true;
        this.client.close();
    }


    public static void main(String[] args) throws InterruptedException, SocketException {
        UDPListener t = new UDPListener(0);
        ExecutorService e = Executors.newFixedThreadPool(1);
        t.open();
        e.submit(t);
        Thread.sleep(1000);
        t.stop();
    }
}

我对 Selector 类也有同样的问题。

我做错了什么？

Answer 1

您忘记添加对 ExecutorService#shutdownNow() 的调用方法。

我冒昧地对代码进行了一些重构，并添加了基本的日志记录语句，以便您可以直观地看到代码在哪些方面做得更好。

这是一个示例输出:

javac UDPListener.java && java UDPListener
[main] Started main application.
[main] Sleeping for 2000
[pool-1-thread-1] Starting...
[pool-1-thread-1] Opening datagram socket...
[pool-1-thread-1] Opened datagram socket.
[pool-1-thread-1] Waiting for datagrams...
[main] Slept for 2000
[main] Stopping all threads.
[pool-1-thread-1] Interrupted while waiting for datagrams.
[pool-1-thread-1] Stopped.
[main] Stopped main application.

固定代码如下:

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class UDPListener implements Runnable {

    private volatile boolean isRunning;
    private volatile boolean clientClosed;
    private int port;
    private DatagramSocket client;

    public UDPListener(int listeningPort) {
        this.port = listeningPort;
    }

    private void open() throws SocketException {
        System.out.println(getName() + " Opening datagram socket...");
        this.clientClosed = false;
        if (this.client != null) {
            this.client.close();
        }
        this.client = new DatagramSocket(this.port);
        System.out.println(getName() + " Opened datagram socket.");
    }

    @Override
    public void run() {
        System.out.println(getName() + " Starting...");
        try {
            this.open();

            this.isRunning = true;
            this.clientClosed = false;
            final DatagramPacket packet = new DatagramPacket(new byte[256], 256);

            while (this.isRunning) {
                System.out.println(getName() + " Waiting for datagrams...");
                try {
                    this.client.receive(packet);
                } catch (SocketException e) {
                    if (!this.clientClosed) {
                        e.printStackTrace();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(getName() + " Interrupted while waiting for datagrams.");
                }
            }
        } catch (SocketException se) {
            se.printStackTrace();
        } finally {
            this.client.disconnect();
            System.out.println(getName() + " Stopped.");
        }
    }

    public void stop() {
        this.isRunning = false;
        this.clientClosed = true;
        this.client.close();
    }

    private static String getName() {
        return "[" + Thread.currentThread().getName() + "]";
    }


    public static void main(String[] args) throws InterruptedException, SocketException {
        System.out.println(getName() + " Started main application.");
        UDPListener t = new UDPListener(0);
        ExecutorService e = Executors.newFixedThreadPool(1);
        // t.open(); easier to debug if this call is exclusively done by UDPListener#run()
        e.submit(t);
        final int sleep = 2000;
        System.out.printf(getName() + " Sleeping for %d\n", sleep);
        Thread.sleep(sleep);
        System.out.printf(getName() + " Slept for %d\n", sleep);
        System.out.println(getName() + " Stopping all threads.");
        t.stop();
        e.shutdownNow();
        System.out.println(getName() + " Stopped main application.");
    }
}