张开涛：线程的中断、超时与降级

最近一位朋友在公众号留言问一个关于熔断的问题：使用hystrix进行httpclient超时熔断错误，我是顺序操作的(没有并发)，发现hystrix会超时断开，但是会导致hystrix线程池不断增多，直到后面因线程池装不下拒绝?

专注于为中小企业提供成都做网站、成都网站建设服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业巴东免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了超过千家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。

而该问题跟线程中断、超时与降级等有关，因此本文将详细介绍导致这个问题背后的原因。

当我们在线程中执行如用户请求任务时，比如HTTP处理线程，最担心的什么?

  1. 线程数***增长;
  2. 线程执行时间长;
  3. 线程不可中断。

对于线程数***增长，我们可以通过使用线程池来控制线程数量，控制线程不是***增长的。

对于线程执行时间长，我们应设置合理的超时时间来保障线程执行时间可控，当超时时要么返回给用户错误页面，要么可以返回降级页面。

对于线程不可中断，我们应想办法将线程设计的可中断，从而在遇到问题可中断线程并降级处理。

接下来的部分将主要讲解线程中断。

线程中断是通过Thread.interrupt()方法来做，一般是在A线程来中断B线程。

首先我们来看下该方法的一些Javadoc描述：

1. 如果线程被Object的wait()、wait(long)、wait(long, int) 或者

Thread的join()、join(long)、join(long, int)、sleep(long)、sleep(long, int)方法阻塞，执行线程中断，且抛出InterruptedException，但中断状态并清空重置，即Thread. isInterrupted()返回false;

2. 如果线程被java.nio.channels.InterruptibleChannel上的一个I/O操作阻塞，执行线程中断，且该

InterruptibleChannel将被关闭，抛出java.nio.channels.ClosedByInterruptException，线程中断状态会设置，即Thread. isInterrupted()返回true;

3. 如果线程被java.nio.channels.Selector阻塞，执行线程中断，该Selector#select()方法将立即返回，相当于调用了java.nio.channels.Selector#wakeup()，不会抛出异常，但会设置中断状态，即Thread. isInterrupted()返回true;

4. 如果不满足以上条件的，那么执行线程中断不会抛出异常，仅设置中断状态，即Thread. isInterrupted()返回true。也就是说我们代码要根据该状态来决定下一步怎么做。

从如上描述可以看出，如果方法异常描述上有抛出

InterruptedException、ClosedByInterruptException异常的，说明该方法可以中断，如“public final native void wait(longtimeout) throws InterruptedException”，但是中断状态会被会被重置要看其Javadoc描述。其他情况基本都是设置中断状态而不会中断掉操作。

BIO(Blocking I/O)操作不可中断

如java.net.Socket读写网络I/O时是阻塞的，除了设置超时时间外，还应该考虑让它可中断或者尽早中断。可以参考《你的Java代码可中断吗》。还有如JDBC驱动mysql-connector-java、HttpClient等大部分都是使用BIO，它们也是不可中断的。

NIO(New I/O)操作可中断

NIO涉及到两部分：

java.nio.channels.Selector和java.nio.channels.InterruptibleChannel，它们是可中断的。如java.nio.channels#SocketChannel实现了InterruptibleChannel，如下方法都是可中断的，并会抛出ClosedByInterruptException异常：

  • connect(SocketAddress remote)
  • read(ByteBuffer[] dsts, int offset, int length)
  • read(ByteBuffer[] dsts)
  • write(ByteBuffer src)

线程、BIO与中断

我们使用BIO实现的HttpClient来做个实验，如下代码所示：

 
 
 
 
    
  
  
  
  
  1. public class BlockingIOTest {
    2. 
  
  
  
  
  2.   public static void main(String[] args) throws Exception {
    3. 
  
  
  
  
  3.     Thread threadA = new Thread(()-> {
    4. 
  
  
  
  
  4.       try {
    5. 
  
  
  
  
  5.        //该阻塞5s
    6. 
  
  
  
  
  6.       String url = "http://localhost:9090/ajax";
    7. 
  
  
  
  
  7.       //HttpClient是BIO，不可中断
    8. 
  
  
  
  
  8.       HttpResponse response = HttpClientUtils.getHttpClient().execute(new HttpGet(url));
    9. 
  
  
  
  
  9.       System.out.println("http status code : " + response.getStatusLine().getStatusCode());
    10. 
  
  
  
  
  10.       //虽然在threadB执行了threadA线程中断
    11. 
  
  
  
  
  11.       //但是仅仅是设置了中断状态为true
    12. 
  
  
  
  
  12.       //并没有中断线程A的执行，该线程还是正常的执行完成了
    13. 
  
  
  
  
  13.       System.out.println("threadA is interrupted: " + Thread.currentThread().isInterrupted());
    14. 
  
  
  
  
  14.      } catch (Exception e) {
    15. 
  
  
  
  
  15.        e.printStackTrace();
    16. 
  
  
  
  
  16.      }
    17. 
  
  
  
  
  17.     });
    18. 
  
  
  
  
  18.     Thread threadB = new Thread(()-> {
    19. 
  
  
  
  
  19.       try {
    20. 
  
  
  
  
  20.        Thread.sleep(2000L);
    21. 
  
  
  
  
  21.       //休眠2s后，中断线程A
    22. 
  
  
  
  
  22.       threadA.interrupt();
    23. 
  
  
  
  
  23.      } catch (Exception e) {
    24. 
  
  
  
  
  24.      }
    25. 
  
  
  
  
  25.     });
    26. 
  
  
  
  
  26.     threadA.start();
    27. 
  
  
  
  
  27.     threadB.start();
    28. 
  
  
  
  
  28.     Thread.sleep(15000L);
    29. 
  
  
  
  
  29.   }
    30. 
  
  
  
  
  30. }
    31.

如上代码的输出结果为：

 
 
 
 
    
  
  
  
  
  1. http status code : 200
    2. 
  
  
  
  
  2. threadA is interrupted: true
    3.

如上代码的执行流程是这样的：

  1. 线程A通过BIO实现HttpClient远程调用http://localhost:9090/ajax获取数据，而该服务需要5s才能响应;
  2. 线程B在线程A执行2s后进行了中断处理，但是线程A调用的HttpClient是阻塞且不可中断的操作，仅仅是设置了线程A的中断状态为true，因此其一直等待网络I/O完成;
  3. 当线程A从远程获取到结果后继续执行，Thread.currentThread().isInterrupted()将输出true，表示线程A被设置了中断状态。

从而需要注意设置了中断状态与中断执行不是一回事。因此对于使用BIO，一定要设置好连接和读写的超时时间，另外可以参考《你的Java代码可中断吗》进行可中断设计。

线程池、Future与中断

我们往线程池提交一个HttpClient任务，并通过Future来等待执行结果，如下代码所示：

 
 
 
 
    
  
  
  
  
  1. public class ThreadPoolTest {
    2. 
  
  
  
  
  2.   private static ExecutorService executorService = Executors.newFixedThreadPool(5);
    3. 
  
  
  
  
  3.   public static void main(String[] args) throws Exception {
    4. 
  
  
  
  
  4.    Future futureA = executorService.submit((Callable) () -> {
    5. 
  
  
  
  
  5.     //该url会阻塞5s
    6. 
  
  
  
  
  6.    String url = "http://localhost:9090/ajax";
    7. 
  
  
  
  
  7.    //HttpClient是BIO，不可中断
    8. 
  
  
  
  
  8.    HttpResponse response = HttpClientUtils.getHttpClient().execute(new HttpGet(url));
    9. 
  
  
  
  
  9.    Integer result = response.getStatusLine().getStatusCode();
    10. 
  
  
  
  
  10.    System.out.println("thread a result : " + result);
    11. 
  
  
  
  
  11.    return response.getStatusLine().getStatusCode();
    12. 
  
  
  
  
  12.    });
    13. 
  
  
  
  
  13.    Future futureB = executorService.submit((Callable) () -> {
    14. 
  
  
  
  
  14.      //该url会阻塞5s
    15. 
  
  
  
  
  15.     String url = "http://localhost:9090/ajax";
    16. 
  
  
  
  
  16.     //HttpClient是BIO，不可中断
    17. 
  
  
  
  
  17.     HttpResponse response = HttpClientUtils.getHttpClient().execute(new HttpGet(url));
    18. 
  
  
  
  
  18.     Integer result = response.getStatusLine().getStatusCode();
    19. 
  
  
  
  
  19.     System.out.println("thread b result : " + result);
    20. 
  
  
  
  
  20.     return result;
    21. 
  
  
  
  
  21.     });
    22. 
  
  
  
  
  22.     try {
    23. 
  
  
  
  
  23.    Integer resultA = futureA.get(100, TimeUnit.MILLISECONDS);
    24. 
  
  
  
  
  24.    } catch (TimeoutException e) {
    25. 
  
  
  
  
  25.     System.out.println("future a timeout");
    26. 
  
  
  
  
  26.   }
    27. 
  
  
  
  
  27.   try {
    28. 
  
  
  
  
  28.     Integer resultB = futureB.get(100, TimeUnit.MILLISECONDS);
    29. 
  
  
  
  
  29.   } catch (TimeoutException e) {
    30. 
  
  
  
  
  30.     System.out.println("future b timeout");
    31. 
  
  
  
  
  31.    }
    32. 
  
  
  
  
  32.   executorService.awaitTermination(10000L, TimeUnit.MILLISECONDS);
    33. 
  
  
  
  
  33.   }
    34. 
  
  
  
  
  34. }
    35.

如上代码的输出结果为：

 
 
 
 
    
  
  
  
  
  1. future a timeout
    2. 
  
  
  
  
  2. future b timeout
    3. 
  
  
  
  
  3. thread a result : 200
    4. 
  
  
  
  
  4. thread b result : 200
    5.

如上代码的执行流程是这样的：

  1. 主线程往线程池提交了两个HttpClient阻塞调用任务，该任务响应时间为5s;
  2. 主线程阻塞在两个带超时的Future等待上，Future在等待线程池任务执行结束，Future的超时时间设置为100ms，所以很快就超时并返回了，主线程继续执行，在《亿级流量》中我们用到了很多这种方法进行并发获取数据和降级或熔断处理;
  3. 线程池中的两个任务其实并没有被中断，还是占用着线程池中的线程，在后台继续执行，直到完成。

从如上可以看出，使用Future时只是在主线程解除了阻塞，并没有连带把线程池任务取消掉，还是占用着线程并阻塞执行。

之前有位同学在公众号后台留言咨询：

使用hystrix进行httpclient超时熔断错误，我是顺序操作的(没有并发)，发现hystrix会超时断开，但是会导致hystrix线程池不断增多，直到后面因线程池装不下拒绝?

看完如上示例，应该能解决该读者的疑惑。虽然熔断了，但是线程中的操作并没有真正的中断，而是还占着线程资源。

接下来我们可以简单看下Future其中的一个实现FutureTask：

超时等待方法get(long timeout, TimeUnit unit)伪代码：

 
 
 
 
    
  
  
  
  
  1. while(true) {
    2. 
  
  
  
  
  2.   if (Thread.interrupted()) {//如果当前线程中断了，处理现场，并抛出中断异常
    3. 
  
  
  
  
  3.     //some code
    4. 
  
  
  
  
  4.    throw new InterruptedException();
    5. 
  
  
  
  
  5.   }
    6. 
  
  
  
  
  6.   //判断剩余休眠时间
    7. 
  
  
  
  
  7.   nanos = deadline - System.nanoTime();
    8. 
  
  
  
  
  8.   if (nanos <= 0L) {//如果没有休眠时间了，则处理线程，并终止执行
    9. 
  
  
  
  
  9.     //some code 
    10. 
  
  
  
  
  10.     return state;
    11. 
  
  
  
  
  11.   }
    12. 
  
  
  
  
  12.   //休眠一段时间，内部实现为UNSAFE.park(false, nanos)
    13. 
  
  
  
  
  13.   LockSupport.parkNanos(this, nanos);
    14. 
  
  
  
  
  14. }
    15.

取消方法cancel(boolean mayInterruptIfRunning)伪代码：

 
 
 
 
    
  
  
  
  
  1. if (mayInterruptIfRunning) {//中断当前线程
    2. 
  
  
  
  
  2.   Thread t = runner;
    3. 
  
  
  
  
  3.   if (t != null)
    4. 
  
  
  
  
  4.     t.interrupt();
    5. 
  
  
  
  
  5. }
    6. 
  
  
  
  
  6. //执行UNSAFE.unpark(thread)唤醒休眠的当前线程
    7. 
  
  
  
  
  7. LockSupport.unpark(t);
    8.

即当我们调用Future#cancel时，是通过唤醒Future所在线程实现，当然实际是比这个要复杂的。

回填结果方法set(V v)伪代码：

 
 
 
 
    
  
  
  
  
  1. //修改Future状态为完成
    2. 
  
  
  
  
  2. //保持v的值，从而Future#get能获取到
    3. 
  
  
  
  
  3. //通过LockSupport.unpark(t)唤醒休眠的线程
    4.

当线程池中的线程执行完成后，是通过Future#set把值设置回Future，从而唤醒休眠的线程，即阻塞在Future#get的等待，然后获取到该结果。

锁与中断

synchronized和ReentrantLock#lock()在获取锁的过程中是不可中断的，假设出现了死锁将一直僵持在那，无法终止阻塞。但我们可以使用可中断的

ReentrantLock#lockInterruptibly()方法或者ReentrantLock#tryLock(long timeout, TimeUnit unit)实现可中断。

总结

在设计高可用系统时，尽量使用线程池，而不是通过每个请求创建一个线程来实现，通过线程池的拒绝策略来优雅的拒绝无法处理的请求。

检查整个请求链路设置合理的超时时间，跟调用方协商合理的SLA、降级限流方案。更长的超时时间意味着出现问题时请求堆积的越多，越可能产生雪崩。

明确知道自己的服务是否可中断，如果不可中断，应该使用线程池和Future实现伪可中断，通过Future配置合理的超时时间，当超时时执行相应的降级策略。也要清楚的知道通过Future只是伪中断，线程池中的任务还是在后台执行，当Future超时后进行重试时，会对调用的服务产生更多的请求，从而造成一种DDos，一定要注意相应的处理策略。

池大小、超时时间和中断没有***的配置策略，要根据自己的场景来动态调整，在系统遇到高并发或者异常时，我们要保护什么，放弃什么，要有权衡。

【本文是专栏作者“张开涛”的原创文章，作者微信公众号：开涛的博客( kaitao-1234567)】

网站标题：张开涛：线程的中断、超时与降级
地址分享：http://www.gawzjz.com/qtweb2/news15/29615.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

定制开发知识

同城分类信息

自贡主机托管　　 　梓潼网站建设　　 　小程序开发　　 　成都小谭网创广告　　 　宜宾网站设计　　 　酒柜书柜鞋柜定制　　 　汕尾凯敏网站　　 　企业网站设计　　 　眉山电信机房托管　　 　四川和正源　　 　成都艺考培训　　 　成都帛金建站　　 　广汉锦华建材　　 　重庆移动网站建设　　 　网络推广　　 　腾讯云香港虚拟主机　　 　成都SEO优化　　 　成都logo设计公司　　 　川银永信　　 　建站定制开发