处理不可中断的阻塞(java并发编程第7章)

末蓝、 2022-07-12 07:05 132阅读 0赞

并不是所有的阻塞都是可中断的, 比如InputStream.read方法. 在检测到输入数据可用, 到达流末尾或者抛出异常前, 该方法一直阻塞. 而且阻塞的时候不会检查中断标记, 所以中断线程无法使read从阻塞状态返回. 但是关闭流可以使得read方法抛出异常, 从而从阻塞状态返回.

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** ReaderThread **extends** Thread \{  
2.      **private** **static** **final** **int** BUFSZ = 1024;  
3.      **private** **final** Socket socket;  
4.      **private** **final** InputStream in;  
5.    
6.      **public** ReaderThread(Socket socket) **throws** IOException \{  
7.          **this**.socket = socket;  
8.          **this**.in = socket.getInputStream();  
9.      \}  
10.   
11.     // 覆盖Thread类的interrupt方法, 加入关闭socket的代码  
12.     // 如果发生中断时, 线程阻塞在read方法上, socket的关闭会导致read方法抛出SocketException, 然后run方法运行完毕  
13.     **public** **void** interrupt() \{  
14.         **try** \{  
15.             socket.close();  
16.         \} **catch** (IOException ignored) \{  
17.         \} **finally** \{  
18.             **super**.interrupt();  
19.         \}  
20.     \}  
21.   
22.     **public** **void** run() \{  
23.         **try** \{  
24.             **byte**\[\] buf = **new** **byte**\[BUFSZ\];  
25.             **while** (**true**) \{  
26.                 **int** count = in.read(buf);  
27.                 **if** (count < 0)  
28.                     **break**;  
29.                 **else** **if** (count > 0)  
30.                     processBuffer(buf, count);  
31.             \}  
32.         \} **catch** (IOException e) \{ /\*  Allow thread to exit  \*/  
33.         \}  
34.     \}  
35.   
36.     **private** **void** processBuffer(**byte**\[\] buf, **int** count) \{  
37.         //...  
38.     \}  
39. \}

如果task并非在自己创建的线程里运行, 而是提交给线程池运行的话, 就无法使用上例的方式处理不可中断阻塞了. 之前有过分析, 对于提交给线程池执行的task, 应该通过Future.cancel方法提前终止task的运行, 所以可以考虑重写Future.cancel方法, 在其中加入关闭socket的操作. Future对象是由submit方法返回的, 其源代码如下:

Java代码   ![收藏代码][icon_star.png]

1.  **public** <T> Future<T> submit(Callable<T> task) \{  
2.      **if** (task == **null**)   
3.          **throw** **new** NullPointerException();  
4.      RunnableFuture<T> ftask = newTaskFor(task);  
5.      execute(ftask);  
6.      **return** ftask;  
7.  \}

可知submit方法返回的Future对象是调用newTaskFor方法获得的:

Java代码   ![收藏代码][icon_star.png]

1.  **protected** <T> RunnableFuture<T> newTaskFor(Callable<T> callable) \{  
2.      **return** **new** FutureTask<T>(callable);  
3.  \}

newTaskFor方法被声明为protected, 所以我们可以通过继承覆盖该方法, 返回自定义的Future对象.

首先将需要覆盖的2个方法定义在接口中:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **interface** CancellableTask<T> **extends** Callable<T> \{   
2.      **void** cancel();   
3.      RunnableFuture<T> newTask();   
4.  \}

然后让task类实现CancellableTask接口:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **abstract** **class** SocketUsingTask<T> **implements** CancellableTask<T> \{  
2.      **private** Socket socket;  
3.    
4.      **protected** **synchronized** **void** setSocket(Socket s) \{  
5.          socket = s;  
6.      \}  
7.    
8.      **public** **synchronized** **void** cancel() \{  
9.          **try** \{  
10.             **if** (socket != **null**)  
11.                 socket.close();  
12.         \} **catch** (IOException ignored) \{  
13.         \}  
14.     \}  
15.   
16.     **public** RunnableFuture<T> newTask() \{  
17.         **return** **new** FutureTask<T>(**this**) \{  
18.             // 定义FutureTask的匿名内部类, 并覆盖cancel方法, 向其中加入关闭socket的操作  
19.             **public** **boolean** cancel(**boolean** mayInterruptIfRunning) \{  
20.                 **try** \{  
21.                     SocketUsingTask.**this**.cancel();  
22.                 \} **finally** \{  
23.                     **return** **super**.cancel(mayInterruptIfRunning);  
24.                 \}  
25.             \}  
26.         \};  
27.     \}  
28. \}

接着继承ThreadPoolExecutor类并覆盖newTaskFor方法, 让该方法返回自定义的FutureTask对象:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** CancellingExecutor **extends** ThreadPoolExecutor \{  
2.      // ...  
3.      **protected** <T> RunnableFuture<T> newTaskFor(Callable<T> callable) \{  
4.          // 如果callable是CancellableTask对象, 那么就返回自定义的FutureTask(通过调用其newTaskFor方法实现)  
5.          **if** (callable **instanceof** CancellableTask)  
6.              **return** ((CancellableTask<T>) callable).newTask();  
7.          **else**  
8.              **return** **super**.newTaskFor(callable);  
9.      \}  
10. \}

测试代码:

Java代码   ![收藏代码][icon_star.png]

1.  **public** **static** **void** main(String\[\] args) \{  
2.      CancellingExecutor executor = **new** CancellingExecutor();  
3.      SocketUsingTask task = **new** SocketUsingTask();  
4.      task.setSocket(**new** Socket("www.baidu.com", 80));  
5.      Future<V> future = executor.submit(task);  
6.      future.cancel(**true**);  
7.  \}

### [处理不可中断的阻塞][Link 1] ###

[icon_star.png]: /images/20220711/aa14f96e75d842bd8fbc2fe7593346ad.png
[Link 1]: http://coolxing.iteye.com/blog/1476289