哥妞 海淀山后有哥妞

如何优雅的停止线程

2021-02-27

文中内容整理自《Java并发编程实战》

如何优雅的停止线程?停止线程池呢?

方法一:使用中断标志

使用 cancelled 标志,当标志状态为true的时候,停止线程。

示例:

public class Test0701CancelThread implements Runnable {

    public static void main(String[] args) {
        Test0701CancelThread cancelThread = new Test0701CancelThread();
        try {
            cancelThread.aSecondOfPrimes();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private final List<BigInteger> primes = new ArrayList<>();

    private volatile boolean cancelled;

    @Override
    public void run() {
        System.out.println("执行线程");
        BigInteger p = BigInteger.ONE;
        while (!cancelled) {
            p = p.nextProbablePrime();
            synchronized (this) {
                primes.add(p);
            }
        }
    }

    public void cancel() {
        System.out.println("取消线程");
        cancelled = true;
    }

    public synchronized List<BigInteger> get() {
        return new ArrayList<>(primes);
    }

    /**
     * 测试方法
     * 素数生成器执行1s之后停止
     *
     * @return
     * @throws InterruptedException
     */
    public List<BigInteger> aSecondOfPrimes() throws InterruptedException {
        Test0701CancelThread cancelThread = new Test0701CancelThread();
        new Thread(cancelThread).start();
        try {
            SECONDS.sleep(1);
        } finally {
            cancelThread.cancel();
        }
        return cancelThread.get();
    }
}

问题:

1、如果调用阻塞方法,可能会导致永远也不会检查取消标志,导致线程不能被终止。

方法二:通过 Thread 类的 interrupt() 方法。

通过 Thread 类的中断方法 interrupt() 方法,中断线程。这是停止线程的最佳实践,它不会马上停止线程,而会先保留必要的内容,然后安全地停止线程。

示例:

public class Test0705CancelThread extends Thread {

    private final BlockingQueue<BigInteger> queue;

    public Test0705CancelThread(BlockingQueue<BigInteger> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            BigInteger p = BigInteger.ONE;
            while (!Thread.currentThread().isInterrupted()) {
                queue.put(p = p.nextProbablePrime());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void cancel() {
        interrupt();
    }
}

注意:

  1. 调用 interrupt 并不意味着立即停止目标线程正在进行的工作,而只是传递了请求中断的消息。
  2. 如果调用 interrupted() 方法返回的值true,可以通过再次调用 interrupted() 恢复中断。
  3. 通常,中断是实现取消的最合理方式。

方法三:通过 Future 来取消线程

Future 类,提供的 cancel() 方法,可以停止 Future 任务。

示例:

public class Test0710FutureCancelThread {

    private static final ScheduledExecutorService cancelExec = Executors.newScheduledThreadPool(5);

    public static void timeRun(final Runnable r, long timeout, TimeUnit unit) {
        Future<?> task = cancelExec.submit(r);
        try {
            task.get();
        } catch (InterruptedException e) {
            // 取消线程
            task.cancel(true);
        } catch (ExecutionException e) {
            // 如果任务已经执行完毕,那么执行取消不会有什么影响
            // 如果任务正在执行,那么将被中断
            task.cancel(true);
        }
    }
}

也是取消线程的正确姿势:通过Future来取消线程。当Future.get抛出 InterruptedException或者TimeoutException时,如果你知道不再需要结果,那么就可以调用Future.cancel来取消任务

方法四:调用 shutdown() 方法,关闭 ExecutorService

线程池的停止,可以调用 ExecutorService 类提供的 shutdown() 方法进行停止。

示例:

public class Test0716CloseExecutorService {

    private static final long TIMEOUT = 1000L;
    private static final TimeUnit UNIT = TimeUnit.MILLISECONDS;
    private final ExecutorService exec = Executors.newScheduledThreadPool(5);

    private final PrintWriter writer;

    public Test0716CloseExecutorService(PrintWriter writer) {
        this.writer = writer;
    }

    public void start() {
    }

    public void stop() throws InterruptedException {
        try {
            exec.shutdown();
            // TODO 问:为什么停止线程,需要调用以下语句
            // 这个方法就是调用shutdown() 之后等待任务执行完毕的方法,可以查看源码的注释
            // Blocks until all tasks have completed execution after a shutdown
            // request, or the timeout occurs, or the current thread is
            // interrupted, whichever happens first.
            exec.awaitTermination(TIMEOUT, UNIT);
        } finally {
            writer.close();
        }
    }

    public void log(String msg) {
        try {
            exec.execute(new WriteTask(msg));
        } catch (RejectedExecutionException ignored) {

        }
    }
}

核心语句如下:

// 不接收新任务
exec.shutdown();
// 停止之前提交的任务
exec.awaitTermination(TIMEOUT, UNIT);

方法五:通过毒丸对象,取消生产者-消费者线程

在生产者-消费者模式中,通过在队列中增加毒丸对象类停止线程。

示例:

public class Test0717PoisonPillCancelThread {

    private static final File POSION = new File("");

    private final IndexerThread consumer = new IndexerThread();

    private final CrawlerThread producer = new CrawlerThread();

    private final BlockingQueue<File> queue;

    private final FileFilter fileFilter;

    private final File root;

    public Test0717PoisonPillCancelThread(BlockingQueue<File> queue, FileFilter fileFilter, File root) {
        this.queue = queue;
        this.fileFilter = fileFilter;
        this.root = root;
    }

    public void start() {
        producer.start();
        consumer.start();
    }

    public void stop() {
        producer.interrupt();
    }

    public void awaitTermination() throws InterruptedException {
        consumer.join();
    }

    /**
     * 消费者
     */
    public class IndexerThread extends Thread {

        @Override
        public void run() {
            try {
                while (true) {
                    File file = queue.take();
                    // 毒丸对象 停止消费者
                    if (file == POSION) {
                        break;
                    } else {
                        indexFile(file);
                    }
                }
            } catch (InterruptedException e) {
                // deal with exception
            } finally {

            }
        }

        private void indexFile(File file) {
        }

        private void crawl(File root) throws InterruptedException {

        }
    }

    /**
     * 生产者
     */
    public class CrawlerThread extends Thread {

        @Override
        public void run() {
            try {
                crawl(root);
            } catch (InterruptedException e) {
                // deal with exception
            } finally {
                while (true) {
                    try {
                        // 队列中存入毒丸
                        queue.put(POSION);
                        break;
                    } catch (InterruptedException e1) {
                        // deal with exception
                    }
                }
            }
        }

        private void crawl(File root) throws InterruptedException {

        }
    }
}

毒丸对象是指放在队列上的对象,其含义是:当得到这个对象时,立即停止。

只有在生产者和消费者的数量都已知的情况下,才可以使用毒丸对象。有多少个生产者,就需要多少个毒丸对象,这样才能停止所有的生产者服务。

代码地址:https://github.com/prepared48/Java-learning.git


1分也是爱


微信扫描二维码,关注公众号 (转载本站文章请注明作者和出处 哥妞-geniu)

Similar Posts

评论