找回密码
 立即注册
首页 业界区 业界 线程如何停止?线程之间如何协作?线程之间的异常如何处 ...

线程如何停止?线程之间如何协作?线程之间的异常如何处理?

辈霖利 5 小时前
线程停止

stop方法

stop 方法虽然可以停止线程,但它已经是不建议使用的废弃方法了,这一点可以通过 Thread 类中的源码发现,stop 源码如下:
1.gif

stop 方法是被 @Deprecated 修饰的不建议使用的过期方法,并且在注释的第一句话就说明了 stop 方法为非安全的方法。
原因在于它在终止一个线程时会强制中断线程的执行,不管run方法是否执行完了,并且还会释放这个线程所持有的所有的锁对象。这一现象会被其它因为请求锁而阻塞的线程看到,使他们继续向下执行。这就会造成数据的不一致。
比如银行转账,从A账户向B账户转账500元,这一过程分为三步,第一步是从A账户中减去500元,假如到这时线程就被stop了,那么这个线程就会释放它所取得锁,然后其他的线程继续执行,这样A账户就莫名其妙的少了500元而B账户也没有收到钱。这就是stop方法的不安全性。
设置标志位

如果线程的run方法中执行的是一个重复执行的循环,可以提供一个标记来控制循环是否继续
  1. class FlagThread extends Thread {
  2.     // 自定义中断标识符
  3.     public volatile boolean isInterrupt = false;
  4.     @Override
  5.     public void run() {
  6.         // 如果为 true -> 中断执行
  7.         while (!isInterrupt) {
  8.             // 业务逻辑处理
  9.         }
  10.     }
  11. }
复制代码
但自定义中断标识符的问题在于:线程中断的不够及时。因为线程在执行过程中,无法调用 while(!isInterrupt) 来判断线程是否为终止状态,它只能在下一轮运行时判断是否要终止当前线程,所以它中断线程不够及时,比如以下代码:
  1. class InterruptFlag {
  2.     // 自定义的中断标识符
  3.     private static volatile boolean isInterrupt = false;
  4.     public static void main(String[] args) throws InterruptedException {
  5.         // 创建可中断的线程实例
  6.         Thread thread = new Thread(() -> {
  7.             while (!isInterrupt) { // 如果 isInterrupt=true 则停止线程
  8.                 System.out.println("thread 执行步骤1:线程即将进入休眠状态");
  9.                 try {
  10.                     // 休眠 1s
  11.                     Thread.sleep(1000);
  12.                 } catch (InterruptedException e) {
  13.                     e.printStackTrace();
  14.                 }
  15.                 System.out.println("thread 执行步骤2:线程执行了任务");
  16.             }
  17.         });
  18.         thread.start(); // 启动线程
  19.         // 休眠 100ms,等待 thread 线程运行起来
  20.         Thread.sleep(100);
  21.         System.out.println("主线程:试图终止线程 thread");
  22.         // 修改中断标识符,中断线程
  23.         isInterrupt = true;
  24.     }
  25. }
复制代码
输出:我们期望的是:线程执行了步骤 1 之后,收到中断线程的指令,然后就不要再执行步骤 2 了,但从上述执行结果可以看出,使用自定义中断标识符是没办法实现我们预期的结果的,这就是自定义中断标识符,响应不够及时的问题。
2.gif

interrupted中断

这种方式需要在while循环中判断使用
使用 interrupt 方法可以给执行任务的线程,发送一个中断线程的指令,它并不直接中断线程,而是发送一个中断线程的信号,把是否正在中断线程的主动权交给代码编写者。相比于自定义中断标识符而然,它能更及时的接收到中断指令,如下代码所示:
  1. public static void main(String[] args) throws InterruptedException {
  2.     // 创建可中断的线程实例
  3.     Thread thread = new Thread(() -> {
  4.         while (!Thread.currentThread().isInterrupted()) {
  5.             System.out.println("thread 执行步骤1:线程即将进入休眠状态");
  6.             try {
  7.                 // 休眠 1s
  8.                 Thread.sleep(1000);
  9.             } catch (InterruptedException e) {
  10.                 System.out.println("thread 线程接收到中断指令,执行中断操作");
  11.                 // 中断当前线程的任务执行
  12.                 break;
  13.             }
  14.             System.out.println("thread 执行步骤2:线程执行了任务");
  15.         }
  16.     });
  17.     thread.start(); // 启动线程
  18.     // 休眠 100ms,等待 thread 线程运行起来
  19.     Thread.sleep(100);
  20.     System.out.println("主线程:试图终止线程 thread");
  21.     // 修改中断标识符,中断线程
  22.     thread.interrupt();
  23. }
复制代码
输出:
3.gif

从上述结果可以看出,线程在接收到中断指令之后,立即中断了线程,相比于上一种自定义中断标识符的方法来说,它能更及时的响应中断线程指令。
利用interruptedException

这种方式 不 需要在while循环中判断使用
如果线程因为执行join(),sleep或者wait()而进入阻塞状态,此时想要停止它,可以调用interrupt(),程序会抛出interruptedException异常。可以利用这个异常终止线程
  1. public void run() {
  2.     System.out.println(this.getName() + "start");
  3.     int i=0;
  4.     //while (!Thread.interrupted()){
  5.     while (!Thread.currentThread().isInterrupted()){
  6.         try {
  7.             Thread.sleep(10000);
  8.         } catch (InterruptedException e) {
  9.             //e.printStackTrace();
  10.             System.out.println("中断线程");
  11.             break;//通过识别到异常来中断
  12.         }
  13.         System.out.println(this.getName() + " "+ i);
  14.         i++;
  15.     }
  16.     System.out.println(this.getName() + "end");
  17. }
复制代码
4.gif

Executor 的中断操作

调用 Executor 的 shutdown() 方法会等待线程都执行完毕之后再关闭,但是如果调用的是 shutdownNow() 方法,则相当于调用每个线程的 interrupt() 方法。
以下使用 Lambda 创建线程,相当于创建了一个匿名内部线程。
  1. public static void main(String[] args) {
  2.     ExecutorService executorService = Executors.newCachedThreadPool();
  3.     executorService.execute(() -> {
  4.         try {
  5.             Thread.sleep(2000);
  6.             System.out.println("Thread run");
  7.         } catch (InterruptedException e) {
  8.             e.printStackTrace();
  9.         }
  10.     });
  11.     executorService.shutdownNow();
  12.     System.out.println("Main run");
  13. }
复制代码
  1. Main run
  2. java.lang.InterruptedException: sleep interrupted
  3.     at java.lang.Thread.sleep(Native Method)
  4.     at ExecutorInterruptExample.lambda$main$0(ExecutorInterruptExample.java:9)
  5.     at ExecutorInterruptExample$$Lambda$1/1160460865.run(Unknown Source)
  6.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  7.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  8.     at java.lang.Thread.run(Thread.java:745)
复制代码
如果只想中断 Executor 中的一个线程,可以通过使用 submit() 方法来提交一个线程,它会返回一个 Future 对象,通过调用该对象的 cancel(true) 方法就可以中断线程。
  1. Future<?> future = executorService.submit(() -> {
  2.     // ..
  3. });
  4. future.cancel(true);
复制代码
线程之间的协作

当多个线程可以一起工作去解决某个问题时,如果某些部分必须在其它部分之前完成,那么就需要对线程进行协调。
join()

案例

在线程中调用另一个线程的 join() 方法,会将当前线程挂起,而不是忙等待,直到目标线程结束。
对于以下代码,虽然 b 线程先启动,但是因为在 b 线程中调用了 a 线程的 join() 方法,b 线程会等待 a 线程结束才继续执行,因此最后能够保证 a 线程的输出先于 b 线程的输出。
  1. public class JoinExample {
  2.     private class A extends Thread {
  3.         @Override
  4.         public void run() {
  5.             System.out.println("A");
  6.         }
  7.     }
  8.     private class B extends Thread {
  9.         private A a;
  10.         B(A a) {
  11.             this.a = a;
  12.         }
  13.         @Override
  14.         public void run() {
  15.             try {
  16.                 a.join();
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.             System.out.println("B");
  21.         }
  22.     }
  23.     public void test() {
  24.         A a = new A();
  25.         B b = new B(a);
  26.         b.start();
  27.         a.start();
  28.     }
  29. }
复制代码
  1. public static void main(String[] args) {
  2.     JoinExample example = new JoinExample();
  3.     example.test();
  4. }
复制代码
  1. A
  2. B
复制代码
原理

5.gif
  1. public final synchronized void join(long millis)
  2. throws InterruptedException {
  3.     long base = System.currentTimeMillis();
  4.     long now = 0;
  5.     if (millis < 0) {
  6.         throw new IllegalArgumentException("timeout value is negative");
  7.     }
  8.     if (millis == 0) {
  9.         while (isAlive()) {//检查线程是否存活,只要线程还没结束,主线程就会一直阻塞
  10.             wait(0);//这里的wait调用的本地方法。
  11.         }
  12.     } else {//等待一段指定的时间
  13.         while (isAlive()) {
  14.             long delay = millis - now;
  15.             if (delay <= 0) {
  16.                 break;
  17.             }
  18.             wait(delay);
  19.             now = System.currentTimeMillis() - base;
  20.         }
  21.     }
  22. }
复制代码
其实是jvm虚拟机中存在方法lock.notify_all(thread),在t1线程结束以后,会调用该方法,最后唤醒主线程。
所以简化一下,流程即:
6.gif

wait() notify() notifyAll()

调用 wait() 使得线程等待某个条件满足,线程在等待时会被挂起,当其他线程的运行使得这个条件满足时,其它线程会调用 notify() 或者 notifyAll() 来唤醒挂起的线程。
它们都属于 Object 的一部分,而不属于 Thread。
只能用在同步方法synchronized或者同步控制块中使用,否则会在运行时抛出 IllegalMonitorStateExeception。
使用 wait() 挂起期间,线程会释放锁。这是因为,如果没有释放锁,那么其它线程就无法进入对象的同步方法或者同步控制块中,那么就无法执行 notify() 或者 notifyAll() 来唤醒挂起的线程,造成死锁。
  1. public final void join(long millis){
  2. synchronized(this){
  3.         //代码块
  4.     }
  5. }
复制代码
  1. //t1.join()前的代码
  2. synchronized (t1) {
  3. // 调用者线程进入 t1 的 waitSet 等待, 直到 t1 运行结束
  4. while (t1.isAlive()) {
  5.   t1.wait(0);
  6. }
  7. }
  8. //t1.join()后的代码
复制代码
  1. static void ensure_join(JavaThread* thread) {
  2. Handle threadObj(thread, thread->threadObj());
  3. ObjectLocker lock(threadObj, thread);
  4. hread->clear_pending_exception();
  5. //这一句中的TERMINATED表示这是线程结束以后运行的
  6. java_lang_Thread::set_thread_status(threadObj(), java_lang_Thread::TERMINATED);
  7.     //这里会清楚native线程,isAlive()方法会返回false
  8.     java_lang_Thread::set_thread(threadObj(), NULL);
  9. //thread就是当前线程,调用这个方法唤醒等待的线程。
  10. lock.notify_all(thread);
  11. hread->clear_pending_exception();
  12. }
复制代码
wait() 和 sleep() 的区别

  • wait() 是 Object 的方法,而 sleep() 是 Thread 的静态方法;
  • wait() 会释放锁,sleep() 不会。
await() signal() signalAll()

java.util.concurrent 类库中提供了 Condition 类来实现线程之间的协调,可以在 Condition 上调用 await() 方法使线程等待,其它线程调用 signal() 或 signalAll() 方法唤醒等待的线程。相比于 wait() 这种等待方式,await() 可以指定等待的条件,因此更加灵活。
使用 Lock 来获取一个 Condition 对象。
  1. public class WaitNotifyExample {
  2.     public synchronized void before() {
  3.         System.out.println("before");
  4.         notifyAll();
  5.     }
  6.     public synchronized void after() {
  7.         try {
  8.             wait();
  9.         } catch (InterruptedException e) {
  10.             e.printStackTrace();
  11.         }
  12.         System.out.println("after");
  13.     }
  14. }
复制代码
  1. public static void main(String[] args) {
  2.     ExecutorService executorService = Executors.newCachedThreadPool();
  3.     WaitNotifyExample example = new WaitNotifyExample();
  4.     executorService.execute(() -> example.after());
  5.     executorService.execute(() -> example.before());
  6. }
复制代码
  1. static void ensure_join(JavaThread* thread) {
  2. Handle threadObj(thread, thread->threadObj());
  3. ObjectLocker lock(threadObj, thread);
  4. hread->clear_pending_exception();
  5. //这一句中的TERMINATED表示这是线程结束以后运行的
  6. java_lang_Thread::set_thread_status(threadObj(), java_lang_Thread::TERMINATED);
  7.     //这里会清楚native线程,isAlive()方法会返回false
  8.     java_lang_Thread::set_thread(threadObj(), NULL);
  9. //thread就是当前线程,调用这个方法唤醒等待的线程。
  10. lock.notify_all(thread);
  11. hread->clear_pending_exception();
  12. }
复制代码
线程中的异常处理

Runnable中异常如何被吞掉

Runnable 接口的 run() 方法不允许抛出任何被检查的异常(checked exceptions),只能处理或抛出运行时异常(unchecked exceptions)。当在 run() 方法内发生异常时,如果没有显式地捕获和处理这些异常,它们通常会在执行该 Runnable 的线程中被“吞掉”,即异常会导致线程终止,但不会影响其他线程的执行。
  1. public class AwaitSignalExample {
  2.     private Lock lock = new ReentrantLock();
  3.     private Condition condition = lock.newCondition();
  4.     public void before() {
  5.         lock.lock();
  6.         try {
  7.             System.out.println("before");
  8.             condition.signalAll();
  9.         } finally {
  10.             lock.unlock();
  11.         }
  12.     }
  13.     public void after() {
  14.         lock.lock();
  15.         try {
  16.             condition.await();
  17.             System.out.println("after");
  18.         } catch (InterruptedException e) {
  19.             e.printStackTrace();
  20.         } finally {
  21.             lock.unlock();
  22.         }
  23.     }
  24. }
复制代码
解决方案:

  • 在run方法中显示的捕获异常
    1. public static void main(String[] args) {
    2.     ExecutorService executorService = Executors.newCachedThreadPool();
    3.     AwaitSignalExample example = new AwaitSignalExample();
    4.     executorService.execute(() -> example.after());
    5.     executorService.execute(() -> example.before());
    6. }
    复制代码
  • 为创建的线程设置一个UncaughtExceptionHandler
    1. before
    2. after
    复制代码
  • 使用Callable代替Runnable,Callable的call方法允许抛出异常,然后可以通过提交给ExecutorService返回的Future来捕获和处理这些异常
    1. public void uncaughtException(Thread t, Throwable e) {
    2.    if (parent != null) {
    3.         parent.uncaughtException(t, e);
    4.    } else {
    5.         Thread.UncaughtExceptionHandler ueh =
    6.             Thread.getDefaultUncaughtExceptionHandler();
    7.         if (ueh != null) {
    8.             ueh.uncaughtException(t, e);
    9.         } else if (!(e instanceof ThreadDeath)) {
    10.             System.err.print("Exception in thread ""
    11.                              + t.getName() + "" ");
    12.             e.printStackTrace(System.err);
    13.         }
    14.     }
    15. }
    复制代码
Callable中异常如何被吞掉
  1. public void run() {
  2.     try {
  3.         // 可能抛出异常的代码
  4.     } catch (Exception e) {
  5.         // 记录日志或处理异常
  6.         throw new RuntimeException(e);
  7.     }
  8. }
复制代码
运行结果
  1. Thread t = new Thread(() -> {
  2.    int i = 1 / 0;
  3. }, "t1");
  4. t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
  5.    @Override
  6.    public void uncaughtException(Thread t, Throwable e) {
  7.         logger.error('---', e);
  8.    }
  9. });
复制代码
源码如下:
  1. ExecutorService executor = Executors.newFixedThreadPool(1);
  2. Future<?> future = executor.submit(() -> {
  3.     // 可能抛出异常的代码
  4. });
  5. try {
  6.     future.get(); // 这里会捕获到Callable中的异常
  7. } catch (ExecutionException e) {
  8.     Throwable cause = e.getCause(); // 获取原始异常
  9. }
复制代码
RunableFuture 是个接口,但是它继承了Runnable 接口 , 实现类是 FutureTask ,因此就需要看下 FutureTask里的run方法 是不是和 构造时的Callable 有关系:
  1. class MyCallable implements Callable<String> {
  2.     @Override
  3.     public String call() throws Exception {
  4.         System.out.println("===> 开始执行callable");
  5.         int i = 1 / 0; //异常的地方
  6.         return "callable的结果";
  7.     }
  8. }
  9. public class CallableAndRunnableTest {
  10.     public static void main(String[] args) {
  11.         System.out.println(" =========> main start ");
  12.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
  13.         Future<String> submit = threadPoolExecutor.submit(new MyCallable());
  14.         try {
  15.             TimeUnit.SECONDS.sleep(2);
  16.         } catch (InterruptedException e) {
  17.             e.printStackTrace();
  18.         }
  19.         System.out.println(" =========> main end ");
  20.     }
  21. }
复制代码
接下来就要看,如果存储正常结果的set(result)方法 和存储异常结果的 setException(ex) 方法
  1. =========> main start
  2. ===> 开始执行callable
  3. =========> main end
复制代码
这两个代码都做了一个操作,就是将正常结果result 和 异常结果 exception 都赋值给了 outcome 这个变量 。
接着再看下future的get方法
  1. public <T> Future<T> submit(Callable<T> task) {
  2.     if (task == null) throw new NullPointerException();
  3.     RunnableFuture<T> ftask = newTaskFor(task);
  4.     execute(ftask);
  5.     return ftask;
  6. }
  7. protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
  8.     return new FutureTask<T>(callable);
  9. }
复制代码
因此可以通过get方法获取到异常结果

来源:程序园用户自行投稿发布,如果侵权,请联系站长删除
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!

相关推荐

您需要登录后才可以回帖 登录 | 立即注册