Java 可中断线程

2023-02-12 20:43:34

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. System.out.println("main: start thread.");
  9. th.start();
  10. // 等待2秒
  11. Thread.sleep(2000);
  12. // 中断线程
  13. System.out.println("main: interrupt thread.");
  14. th.interrupt();
  15. // 等待2秒
  16. Thread.sleep(2000);
  17. // 停止线程
  18. System.out.println("main: stop thread.");
  19. th.stop();
  20. }
  21. private static class TestRunnable implements Runnable
  22. {
  23. @Override
  24. public void run()
  25. {
  26. System.out.println("Thread started.");
  27. while (true)
  28. {
  29. // 避免由于while(true)导致编译失败。
  30. if (false) break;
  31. }
  32. // 清理工作
  33. System.out.println("Thread ended.");
  34. }
  35. }
  36. }
  • 输出结果：

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

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

例子中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)
  • 不能设置超时也不能抛出异常的阻塞方法：
  • synchronized块，Object的wait方法。可以使用ReentrantLock和Condition替代。
  1. 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. }

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. Thread.sleep(500);
  12. }
  13. }
  14. catch (InterruptedException e)
  15. {
  16. }
  17. }

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. }
  16. // 清理操作
  17. }

condition.await();

  • // 其他操作
  • locker.unlock();
  • }
  • }
  • catch (InterruptedException e)
  • {
  • }
  • // 清理操作
  • }

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. }
  16. private void produce() throws InterruptedException
  17. {
  18.     locker.lockInterruptibly();
  19.     try
  20.     {
  21.         // Produce
  22.         int x = (int)(Math.random() * 100);
  23.         queue.offer(x);
  24.         emptyCondition.signalAll();
  25.         System.out.printf("producer: %d and signal all. queue = %d /n",
  26.                 x, queue.size());
  27.     }
  28.     finally
  29.     {
  30.         locker.unlock();
  31.     }
  32. }
  • Consumer线程被中断后不结束，直到队列内所有数据被输出。
  • 不使用Thread.currentThread().isInterrupted()而定义isInt记录中断，可以避免中断导致ReentrantLock方法的tryLock不能获得锁而直接抛出异常。
  1. public void run()
  2. {
  3. // 线程是否被中断
  4. boolean isInt = false;
  5. // 线程中断后，将队列内的数据全部读出，再结束线程。
  6. while (!isInt || !queue.isEmpty())
  7. {
  8. try
  9. {
  10. consume();
  11. Thread.sleep((int)(Math.random() * 3000));
  12. }
  13. catch (InterruptedException e)
  14. {
  15. isInt = true;
  16. }
  17. }
  18. System.out.println("consumer: end.");
  19. }
  20. private void consume() throws InterruptedException
  21. {
  22. if (!locker.tryLock(5, TimeUnit.SECONDS))
  23. {
  24. // 没有获得锁,不进行任何操作。 避免死锁。
  25. return;
  26. }
  27. try
  28. {
  29. // Consume
  30. while (queue.isEmpty())
  31. {
  32. System.out.println("consumer: waiting for condition.");
  33. if (!emptyCondition.await(5, TimeUnit.SECONDS))
  34. {
  35. // 5秒没有等待到条件，不进行任何操作。
  36. // 避免中断后在此处等待。
  37. return;
  38. }
  39. }
  40. int x = queue.poll();
  41. System.out.printf("consumer: %d, queue=%d /n", x, queue.size());
  42. }
  43. finally
  44. {
  45. locker.unlock();
  46. }
  47. }

queue.put(x);

  • System.out.println("producer: " + x);
  • Thread.sleep((int)(Math.random() * 3000));
  • }
  • }
  • catch (InterruptedException e)
  • {
  • }
  • System.out.println("producer: end.");
  • }

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. while (!Thread.currentThread().isInterrupted())
  14. {
  15. try
  16. {
  17. @SuppressWarnings("unused")
  18. Socket s = serverSocket.accept();
  19. }
  20. catch (SocketTimeoutException e)
  21. {
  22. // 超时，不进行任何处理，再次调用accept方法
  23. }
  24. catch (IOException e)
  25. {
  26. e.printStackTrace();
  27. }
  28. }
  29. // 清理工作
  30. }

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

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

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

  • 与readLine的处理方法类似。
  • 调用Socket的setSoTimeout方法设置超时时间。
  • ObjectInputStream的readObject方法在阻塞指定的时间后将抛出异常。
  • 不需要处理该异常。
  1. public void run()
  2. {
  3.     ObjectInputStream ois = null;
  4.     try
  5.     {
  6.         // 建立测试用的链接
  7.         serverSocket = new ServerSocket(10009);
  8.         socket = new Socket("127.0.0.1", 10009);
  9.         Thread.sleep(500);
  10.         Socket s = serverSocket.accept();
  11.         // 向socket发送3个对象
  12.         ObjectOutputStream oos = new ObjectOutputStream(
  13.                 s.getOutputStream());
  14.         oos.writeObject(new TestData("object 1"));
  15.         oos.writeObject(new TestData("object 2"));
  16.         oos.writeObject(new TestData("object 3"));
  17.         // 设置超时
  18.         socket.setSoTimeout(1000);
  19.         // 创建ObjectInputStream
  20.         ois = new ObjectInputStream(socket.getInputStream());
  21.     }
  22.     catch (IOException e)
  23.     {
  24.         e.printStackTrace();
  25.         System.exit(-1);
  26.     }
  27.     catch (InterruptedException e)
  28.     {
  29.         e.printStackTrace();
  30.         System.exit(-1);
  31.     }
  32.     while (!Thread.currentThread().isInterrupted())
  33.     {
  34.         try
  35.         {
  36.             TestData s = (TestData)ois.readObject();
  37.             if (s == null)
  38.             {
  39.                 // 流结束
  40.                 break;
  41.             }
  42.             // 输出读取的数据
  43.             System.out.println("thread: " + s.data);
  44.         }
  45.         catch (SocketTimeoutException e)
  46.         {
  47.             System.out.println("thread: socket timeout.");
  48.         }
  49.         catch (IOException e)
  50.         {
  51.             e.printStackTrace();
  52.         }
  53.         catch (ClassNotFoundException e)
  54.         {
  55.             e.printStackTrace();
  56.         }
  57.     }
  58.     // 清理
  59.     try { serverSocket.close(); } catch (Exception e) { }
  60.     try { socket.close(); } catch (Exception e) { }
  61.     System.out.println("thread: end.");
  62. }

其中，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
处理阻塞”。

上一篇：Docker：从引擎和运行框架理解Docker（3）


下一篇：SQL Server中 ldf 文件过大的解决方法