Java 可中断线程

http://blog.csdn.net/sapphiron/article/details/3018053

 

PART.1 无法中断的线程

一个无法中断的线程的例子。

1. public   class  UninterruptableThread
2. {
3.      @SuppressWarnings ( "deprecation" )
4.      public   static   void  main(String[] args)  throws  Exception
5.     {
6.         Thread th =  new  Thread( new  TestRunnable());
7.         
8.          // 启动线程
9.         System.out.println( "main: start thread." );
10.         th.start();
11.         
12.          // 等待2秒
13.         Thread.sleep( 2000 );
14.         
15.          // 中断线程
16.         System.out.println( "main: interrupt thread." );
17.         th.interrupt();
18.         
19.          // 等待2秒
20.         Thread.sleep( 2000 );
21.         
22.          // 停止线程
23.         System.out.println( "main: stop thread." );
24.         th.stop();
25.     }
26.     
27.      private   static   class  TestRunnable  implements  Runnable
28.     {
29.          @Override
30.          public   void  run()
31.         {
32.             System.out.println( "Thread started." );
33.              while  ( true )
34.             {
35.                  // 避免由于while(true)导致编译失败。
36.                  if  ( false )  break ;
37.             }
38.             
39.              // 清理工作
40.             System.out.println( "Thread ended." );
41.         }
42.     }
43. }

• 输出结果：

main: start thread.
Thread started.
main: interrupt thread.
main: stop thread.

• Thread对象的interrupt方法仅仅把该线程的一个标志置为true，该方法本身并不包含任何中断线程的操作。
• stop方法可以将线程中止，但通过输出的结果可以发现，”Thread ended”并没有被输出，即线程本身不能进行任何清理工作。

PART.2 可中断的线程

• 线程应不断检查isInterrupted是否为true，当其返回true时，应开始清理并结束线程。（Test1中的while循环）
• Thread.sleep方法会在线程被中断时抛出InterruptedException，线程可以捕获该异常并开始清理和结束线程。（Test2中的Thread.sleep()）
• 如果循环中不时调用Thread.sleep，可以处理isInterrupted。

可中断线程的例子：

1. public   class  GeneralTest
2. {
3.      public   static   void  main(String[] args)  throws  Exception
4.     {
5.         Thread th1 =  new  Thread( new  Test1());
6.         Thread th2 =  new  Thread( new  Test2());
7.         
8.          // 启动 Test1 和 Test2 线程。
9.         System.out.println( "main: starting 'Test1' and 'Test2'." );
10.         th1.start();
11.         th2.start();
12.         
13.          // 等待3秒。
14.         System.out.println( "main: sleeping for 3 seconds." );
15.         Thread.sleep( 3000 );
16.         
17.          // 中断 Test1 和 Test2 线程。
18.         System.out.println( "main: interrupting 'Test1' and 'Test2'." );
19.         th1.interrupt();
20.         th2.interrupt();
21.         
22.          // 等待 Test1 和 Test2 线程结束。
23.         System.out.println( "main: waiting for 'Test1' and 'Test2' to end." );
24.         th1.join();
25.         th2.join();
26.         
27.         System.out.println( "main: end." );
28.     }
29.     
30.      private   static   class  Test1  implements  Runnable
31.     {
32.          @Override
33.          public   void  run()
34.         {
35.             System.out.println( "Test1: start." );
36.              while  (!Thread.currentThread().isInterrupted())
37.             {
38.                  // 其他操作...
39.                 System.out.print( "" );
40.             }
41.             System.out.println( "Test1: end." );
42.         }
43.     }
44.     
45.      private   static   class  Test2  implements  Runnable
46.     {
47.          @Override
48.          public   void  run()
49.         {
50.             System.out.println( "Test2: start." );
51.              try
52.             {
53.                  while  ( true )
54.                 {
55.                      // 其他操作... 
56.                     Thread.sleep( 1000 );
57.                 }
58.             }
59.              catch  (InterruptedException e)
60.             {
61.                 System.out.println( "Test2: InterruptedException" );
62.             }
63.             System.out.println( "Test2: end." );
64.         }
65.     }
66. }

PART.3 isInterrupted()和interrputed()方法的区别

• isInterrupted方法是实例方法，interrupted方法是静态方法。

Thread.currentThread().isInterrupted()
Thread.interrupted()

• interrupted方法在调用之后会将中断标志置为false。在只对线程调用一次interrupt的前提下interrupted方法只会返回一次true。
• 使用interrupted方法判断应确保在判断之后开始结束线程。

isInterrupted和interrupted方法比较的例子：

1. public   class  InterruptedStateTest
2. {
3.      public   static   void  main(String[] args)  throws  Exception
4.     {
5.          // "Test1"
6.         Thread th1 =  new  Thread( new  Test1());
7.         
8.          // 启动 Test1 线程，并在3秒之后中断该线程。
9.         th1.start();
10.         Thread.yield();
11.         
12.         System.out.println( "Test1 started... Waiting 3 seconds." );
13.         Thread.sleep( 3000 );
14.         
15.         System.out.println( "Interrupting Test1..." );
16.         th1.interrupt();
17.         
18.         Thread.sleep( 1000 );
19.         
20.         System.out.println( "---------------------------------------" );
21.                 
22.          // “Test2"
23.         Thread th2 =  new  Thread( new  Test2());
25.          // 启动 Test2 线程，并在3秒之后中断该线程。
26.         th2.start();
27.         Thread.yield();
28.         
29.         System.out.println( "Test2 started... Waiting 3 seconds." );
30.         Thread.sleep( 3000 );
31.         
32.         System.out.println( "Interrupting Test2..." );
33.         th2.interrupt();
34.         
35.         Thread.yield();
36.         
37.          // 主线程结束。
38.         System.out.println( "End of main." );
39.     }
41.     
42.      private   static   class  Test1  implements  Runnable
43.     {
44.          @Override
45.          public   void  run()
46.         {
47.             System.out.println( "Test1 start..." );
48.              while  ( true )
49.             {
50.                  if  (Thread.currentThread().isInterrupted())
51.                 {
52.                      if  (Thread.currentThread().isInterrupted())
53.                     {
54.                         System.out.println( "Interrupted..." );
55.                          break ;
56.                     }
57.                 }
58.             }
59.             System.out.println( "Test1 end..." );
60.         }
61.     }
62.     
63.      private   static   class  Test2  implements  Runnable
64.     {
65.          @Override
66.          public   void  run()
67.         {
68.              // 记录线程开始时间。
69.              long  startTime = System.currentTimeMillis();
70.             
71.             System.out.println( "Test2 start... "  +
72. "Automatically ends in 6 sec." );
73.             
74.              while  ( true )
75.             {
76.                  // 连续判断2次Thread.interrupted()
77.                  if  (Thread.interrupted())
78.                 {
79.                      if  (Thread.interrupted())
80.                     {
81.                         System.out.println( "Interrupted..." );
82.                          break ;
83.                     }
84.                 }
85.                 
86.                  // 如果线程2运行超过6秒将自动结束。
87.                  if  (System.currentTimeMillis() - startTime >  6000  )
88.                 {
89.                     System.out.println( "5 seconds..." );
90.                      break ;
91.                 }
92.             }
93.             System.out.println( "Test2 end" );
94.         }
95.     }
96. }

例子中Test1连续判断2次Thread.currentThread().isInterrupted()， Test1仍然可以正常中断。

1. if  (Thread.currentThread().isInterrupted())
2. {
3.      if  (Thread.currentThread().isInterrupted())
4.     {
5.          // 结束线程。
6.     }
7. }

Test2连续判断2次Thread.interrupted()，因此Test2线程在被调用interrupt之后没有结束。

1. if  (Thread.interrupted())
2. {
3.      if  (Thread.interrupted())
4.     {
5.      // 结束线程。
6.     }
7. }

PART.4 处理阻塞

阻塞操作如BufferedReader的readLine方法，ServerSocket的accept方法将导致线程不能判断 isInterrupted()，因此线程中的阻塞不能永久阻塞。处理阻塞的方法有以下：

• 在调用阻塞方法时设置超时时间：

 

类	方法	超时后的处理
ReentrantLock ReadLock WriteLock	tryLock (long, TimeUnit)	返回 false
Condition	await (long, TimeUnit) awaitNanos (long) awaitUntil (Date)	返回 false
Future	get (long, TimeUnit)	TimeoutException
CyclicBarrier	await (long, TimeUnit)	TimeoutException
CountDownLatch	await (long, TimeUnit)	返回 false
Exchanger	exchange (V, long, TimeUnit)	TimeoutException
Semaphore	tryAcquire (long, TimeUnit)	返回 false
BlockingQueue <E>	offer (E, long, TimeUnit) poll (long, TimeUnit)	返回 false 返回 null
BlockingDeque <E>	offerFirst (E, long, TimeUnit) offerLast (E, long, TimeUnit) poolFirst (long, TimeUnit) poolLast (long, TimeUnit)	返回 false   返回 null
ServerSocket	accept () 通过 setSoTimeout 设置超时时间。	SocketTimeoutException

• 该方法在阻塞时如果线程被中断，可以抛出一个异常：

 

类	方法	异常
Thread	sleep (long) join ()	InterruptedException
ReentrantLock ReadLock WriteLock	lockInterruptibly ()	InterruptedException
Condition	await()	InterruptedException
Future	get ()	InterruptedException
CyclicBarrier	await ()	InterruptedException
CountDownLatch	await ()	InterruptedException
Exchanger	exchange (V)	InterruptedException
Semaphore	acquire ()	InterruptedException
BlockingQueue <E>	put (E) take ()	InterruptedException
BlockingDeque <E>	putFirst (E) putLast (E) takeFirst (E) takeLast (E)	InterruptedException

• 调用不可中断阻塞方法的可中断版本：

 

类	阻塞方法	可中断方法
ReentrantLock ReadLock WriteLock	lock ()	tryLock () tryLock (long, TimeUnit) lockInterruptibly ()
Condition	awaitUninterruptibly ()	await () await (long, TimeUnit) awaitNanos (long) awaitUntil (Date)
Semaphore	acquireUninterruptibly ()	acquire() tryAcquire() tryAcquire(long, TimeUnit)

• 不能设置超时也不能抛出异常的阻塞方法：

1. synchronized块，Object的wait方法。可以使用ReentrantLock和Condition替代。
   
2. BufferedReader的readLine方法，ObjectInputStream得readObject方法。（如果底层流是通过Socket获得，可以通过Socket设置超时）

PART.5 处理Thread.sleep()

1. 捕获异常并结束线程

• 捕获InterruptedException异常，开始清理操作并结束线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.              // 需要进行的操作
8.             Thread.sleep( 500 );
9.         }
10.     }
11.      catch  (InterruptedException e)
12.     {
13.     }
14.     
15.      // 清理操作
16. }

2. 捕获异常，再次调用interrupt

1. public   void  run()
2. {
3.      while  (!Thread.currentThread().isInterrupted())
4.     {
5.          try
6.         {
7.              // 需要进行的操作 1
8.             Thread.sleep( 500 );
9.         }
10.          catch  (InterruptedException e)
11.         {
12.              // 再次调用interrupt
13.             Thread.currentThread().interrupt();
14.         }
15.         
16.          try
17.         {
18.              // 需要进行的操作 2
19.             Thread.sleep( 500 );
20.         }
21.          catch  (InterruptedException e)
22.         {
23.              // 再次调用interrupt
24.             Thread.currentThread().interrupt();
25.         }
26.     }
27.     
28.      // 清理操作
29. }

PART.6 处理ReentrantLock和Condition

1. 通过lockInterruptibly方法中断

• 捕获lockInterruptibly方法可能跑出的InterruptedException，并结束线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.             locker.lockInterruptibly();
8.              // 其他操作
9.             locker.unlock();
10.         }
11.     }
12.      catch  (InterruptedException e)
13.     {
14.     }
15.      // 清理操作
16. }

2. 通过不设置超时的tryLock方法中断

• tryLock方法将不阻塞。
• 通过捕获Thread.sleep的异常（或其他方法）中断线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.              if  (locker.tryLock())
8.             {
9.                  // 其他操作
10.             }
11.             
12.             Thread.sleep( 500 );
13.         }
14.     }
15.      catch  (InterruptedException e)
16.     {
17.     }
18. }

3. 通过设置超时的tryLock方法中断

• 捕获tryLock方法在线程中断时抛出的InterrupedException并结束线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.              if  (locker.tryLock( 1 , TimeUnit.SECONDS))
8.             {
9.                  // 其他操作
10.             }
11.         }
12.     }
13.      catch  (InterruptedException e)
14.     {
15.     }
17.      // 清理操作
18. }

4. 通过捕获Conditon的await方法抛出的异常中断

• 捕获Condition的await方法抛出的异常并结束线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.             locker.lockInterruptibly();
8.             condition.await();
9.              // 其他操作
10.             locker.unlock();
11.         }
12.     }
13.      catch  (InterruptedException e)
14.     {
15.     }
16.      // 清理操作
17. }

5. 可中断的Producer和Consumer示例

• produce方法在线程中断时将跑出InterruptedException。
• run方法捕获该异常，并中断线程。

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.             produce();
8.             Thread.sleep(( int )(Math.random() *  3000 ));
9.         }
10.     }
11.      catch  (InterruptedException e)
12.     {
13.     }
14.     System.out.println( "producer: end." );
15. }
17. private   void  produce()  throws  InterruptedException
18. {
19.     locker.lockInterruptibly();
20.      try
21.     {
22.          // Produce
23.          int  x = ( int )(Math.random() *  100 );
24.         queue.offer(x);
25.         emptyCondition.signalAll();
26.         System.out.printf( "producer: %d and signal all. queue = %d /n" ,
27.                 x, queue.size());
28.     }
29.      finally
30.     {
31.         locker.unlock();
32.     }
33. }

• Consumer线程被中断后不结束，直到队列内所有数据被输出。
• 不使用Thread.currentThread().isInterrupted()而定义isInt记录中断，可以避免中断导致ReentrantLock方法的tryLock不能获得锁而直接抛出异常。

1. public   void  run()
2. {
3.      // 线程是否被中断
4.      boolean  isInt =  false ;
5.     
6.      // 线程中断后，将队列内的数据全部读出，再结束线程。
7.      while  (!isInt || !queue.isEmpty())
8.     {
9.          try
10.         {
11.             consume();
12.             Thread.sleep(( int )(Math.random() *  3000 ));
13.         }
14.          catch  (InterruptedException e)
15.         {
16.             isInt =  true ;
17.         }
18.     }
19.     System.out.println( "consumer: end." );
20. }
22. private   void  consume()  throws  InterruptedException
23. {
24.      if  (!locker.tryLock( 5 , TimeUnit.SECONDS))
25.     {
26.          // 没有获得锁,不进行任何操作。 避免死锁。
27.          return ;
28.     }
29.     
30.      try
31.     {
32.          // Consume
33.          while  (queue.isEmpty())
34.         {
35.             System.out.println( "consumer: waiting for condition." );
36.              if  (!emptyCondition.await( 5 , TimeUnit.SECONDS))
37.             {
38.                  // 5秒没有等待到条件，不进行任何操作。
39.                  // 避免中断后在此处等待。
40.                  return ;
41.             }
42.         }
43.          int  x = queue.poll();
44.         System.out.printf( "consumer: %d, queue=%d /n" , x, queue.size());
45.     }
46.      finally
47.     {
48.         locker.unlock();
49.     }
50. }

PART.7 处理BlockingQueue

• 使用BlockingQueue处理Consumer和Producer问题。
• put和take方法可以在线程被中断时抛出InterruptedException。

Producer:

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.              int  x = ( int )(Math.random() *  100 );
8.             queue.put(x);
9.             System.out.println( "producer: "  + x);
10.             Thread.sleep(( int )(Math.random() *  3000 ));
11.         }
12.     }
13.      catch  (InterruptedException e)
14.     {
15.     }
16.     System.out.println( "producer: end." );
17. }

Consumer:

1. public   void  run()
2. {
3.      try
4.     {
5.          while  ( true )
6.         {
7.              int  x = queue.take();
8.             System.out.println( "consumer: "  + x);
9.             Thread.sleep(( int )(Math.random() *  3000 ));
10.         }
11.     }
12.      catch  (InterruptedException e)
13.     {
14.     }
15.     System.out.println( "consumer: end." );
16. }

PART.8 处理Socket和ServerSocket

1. 处理ServerSocket的accept方法

• 调用ServerSocket的setSoTimeout方法设置超时时间。
• 捕获并处理超时引起的SocketTimeoutException。
• 不需要处理该异常。

1. public   void  run()
2. {
3.      try
4.     {
5.          // 设置超时时间
6.         serverSocket.setSoTimeout( 2000 );
7.     }
8.      catch  (SocketException e)
9.     {
10.         e.printStackTrace();
11.         System.exit(- 1 );
12.     }
13.     
14.      while  (!Thread.currentThread().isInterrupted())
15.     {
16.          try
17.         {
18.              @SuppressWarnings ( "unused" )
19.             Socket s = serverSocket.accept();
20.         }
21.          catch  (SocketTimeoutException e)
22.         {
23.              // 超时，不进行任何处理，再次调用accept方法
24.         }
25.          catch  (IOException e)
26.         {
27.             e.printStackTrace();
28.         }
29.     }
30.      // 清理工作
31. }

2. 处理包装自Socket的BufferedReader的readLine方法

• 调用Socket的setSoTimeout方法设置超时时间。
• BufferedReader的readLine方法在阻塞指定的时间后将抛出SocketTimeoutException。
• 不需要处理该异常。

1. public   void  run()
2. {
3.     BufferedReader reader =  null ;
4.     
5.      try
6.     {
7.          // 建立测试用的链接
8.         serverSocket =  new  ServerSocket( 10009 );
9.         socket =  new  Socket( "127.0.0.1" ,  10009 );
10.         
11.         Thread.sleep( 500 );
12.         Socket s = serverSocket.accept();
13.         
14.          // 向socket发送5行数据
15.         OutputStreamWriter w =  new  OutputStreamWriter(
16.                 s.getOutputStream());
17.         w.write( "line1 /n line2 /n line3 /n line4 /n line5 /n" );
18.         w.flush();
19.         
20.          // 设置超时
21.         socket.setSoTimeout( 1000 );
23.          // 创建BufferedReader
24.         reader =  new  BufferedReader(
25.                  new  InputStreamReader(socket.getInputStream()));
26.     }
27.      catch  (IOException e)
28.     {
29.         e.printStackTrace();
30.         System.exit(- 1 );
31.     }
32.      catch  (InterruptedException e)
33.     {
34.         e.printStackTrace();
35.         System.exit(- 1 );
36.     }
37.     
38.      while  (!Thread.currentThread().isInterrupted())
39.     {
40.          try
41.         {
42.             String s = reader.readLine();
43.             
44.              if  (s ==  null )
45.             {
46.                  // 流结束
47.                  break ;
48.             }
49.             
50.              // 输出读取的数据
51.             System.out.println( "thread: "  + s);
52.         }
53.          catch  (SocketTimeoutException e)
54.         {
55.             System.out.println( "thread: socket timeout." );
56.         }
57.          catch  (IOException e)
58.         {
59.             e.printStackTrace();
60.         }
61.     }
62.     
63.      // 清理
64.      try  { serverSocket.close(); }  catch  (Exception e) { }
65.      try  { socket.close(); }  catch  (Exception e) { }
66.     System.out.println( "thread: end." );
67. }

3. 处理包装自Socket的ObjectInputStream的readObject方法

• 与readLine的处理方法类似。
• 调用Socket的setSoTimeout方法设置超时时间。
• ObjectInputStream的readObject方法在阻塞指定的时间后将抛出异常。
• 不需要处理该异常。

1. public   void  run()
2. {
3.     ObjectInputStream ois =  null ;
5.      try
6.     {
7.          // 建立测试用的链接
8.         serverSocket =  new  ServerSocket( 10009 );
9.         socket =  new  Socket( "127.0.0.1" ,  10009 );
11.         Thread.sleep( 500 );
12.         Socket s = serverSocket.accept();
14.          // 向socket发送3个对象
15.         ObjectOutputStream oos =  new  ObjectOutputStream(
16.                 s.getOutputStream());
17.         oos.writeObject( new  TestData( "object 1" ));
18.         oos.writeObject( new  TestData( "object 2" ));
19.         oos.writeObject( new  TestData( "object 3" ));
21.          // 设置超时
22.         socket.setSoTimeout( 1000 );
24.          // 创建ObjectInputStream
25.         ois =  new  ObjectInputStream(socket.getInputStream());
26.     }
27.      catch  (IOException e)
28.     {
29.         e.printStackTrace();
30.         System.exit(- 1 );
31.     }
32.      catch  (InterruptedException e)
33.     {
34.         e.printStackTrace();
35.         System.exit(- 1 );
36.     }
38.      while  (!Thread.currentThread().isInterrupted())
39.     {
40.          try
41.         {
42.             TestData s = (TestData)ois.readObject();
44.              if  (s ==  null )
45.             {
46.                  // 流结束
47.                  break ;
48.             }
50.              // 输出读取的数据
51.             System.out.println( "thread: "  + s.data);
52.         }
53.          catch  (SocketTimeoutException e)
54.         {
55.             System.out.println( "thread: socket timeout." );
56.         }
57.          catch  (IOException e)
58.         {
59.             e.printStackTrace();
60.         }
61.          catch  (ClassNotFoundException e)
62.         {
63.             e.printStackTrace();
64.         }
65.     }
67.      // 清理
68.      try  { serverSocket.close(); }  catch  (Exception e) { }
69.      try  { socket.close(); }  catch  (Exception e) { }
70.     System.out.println( "thread: end." );
71. }

其中， TestData 是一个简单的可序列化的类。

1. private   static   class  TestData  implements  Serializable
2. {
3.      private   static   final   long  serialVersionUID = 6147773210845607198L;
4.      public  String data;
5.      public  TestData(String data)
6.     {
7.          this .data = data;
8.     }
9. }

 

PART.9 总结

见“PART.2可中断的线程”和“PART.4 处理阻塞”。