文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

并发编程之Future&FutureTask深入解析

2024-12-03 15:49

关注

 前言

Java线程实现方式主要有四种:

其中前两种方式线程执行完后都没有返回值,后两种是带返回值的。

Callable 和 Runnable 接口

Runnable接口

  1. // 实现Runnable接口的类将被Thread执行,表示一个基本的任务 
  2. public interface Runnable { 
  3.     // run方法就是它所有的内容,就是实际执行的任务 
  4.     public abstract void run(); 

 没有返回值

run 方法没有返回值,虽然有一些别的方法也能实现返回值得效果,比如编写日志文件或者修改共享变量等等,但是不仅容易出错,效率也不高。

不能抛出异常

  1. public class RunThrowExceptionDemo { 
  2.  
  3.      
  4.     public void normalMethod() throws IOException { 
  5.         throw new IOException(); 
  6.     } 
  7.  
  8.     class RunnableImpl implements Runnable { 
  9.  
  10.          
  11.         @Override 
  12.         public void run() { 
  13.             try { 
  14.                 throw new IOException(); 
  15.             } catch (IOException e) { 
  16.                 e.printStackTrace(); 
  17.             } 
  18.         } 
  19.     } 
  20.  

 可以看到普通方法 normalMethod 可以在方法签名上抛出异常,这样上层接口就可以捕获这个异常进行处理,但是实现 Runnable 接口的类,run 方法无法抛出 checked Exception,只能在方法内使用 try catch 进行处理,这样上层就无法得知线程中的异常。

设计导致

其实这两个缺陷主要原因就在于 Runnable 接口设计的 run 方法,这个方法已经规定了 run() 方法的返回类型是 void,而且这个方法没有声明抛出任何异常。所以,当实现并重写这个方法时,我们既不能改返回值类型,也不能更改对于异常抛出的描述,因为在实现方法的时候,语法规定是不允许对这些内容进行修改的。

Runnable 为什么设计成这样?

假设 run() 方法可以返回返回值,或者可以抛出异常,也无济于事,因为我们并没有办法在外层捕获并处理,这是因为调用 run() 方法的类(比如 Thread 类和线程池)是 Java 直接提供的,而不是我们编写的。 所以就算它能有一个返回值,我们也很难把这个返回值利用到,而 Callable 接口就是为了解决这两个问题。

Callable接口

  1. public interface Callable { 
  2.     //返回接口,或者抛出异常 
  3.     V call() throws Exception; 

 可以看到 Callable 和 Runnable 接口其实比较相似,都只有一个方法,也就是线程任务执行的方法,区别就是 call 方法有返回值,而且声明了 throws Exception。

Callable 和 Runnable 的不同之处

与 Callable 配合的有一个 Future 接口,通过 Future 可以了解任务执行情况,或者取消任务的执行,还可获取任务执行的结果,这些功能都是 Runnable 做不到的,Callable 的功能要比 Runnable 强大。

Future接口

Future的作用

简单来说就是利用线程达到异步的效果,同时还可以获取子线程的返回值。

比如当做一定运算的时候,运算过程可能比较耗时,有时会去查数据库,或是繁重的计算,比如压缩、加密等,在这种情况下,如果我们一直在原地等待方法返回,显然是不明智的,整体程序的运行效率会大大降低。

我们可以把运算的过程放到子线程去执行,再通过 Future 去控制子线程执行的计算过程,最后获取到计算结果。这样一来就可以把整个程序的运行效率提高,是一种异步的思想。

Future的方法

Future 接口一共有5个方法,源代码如下:

 

  1. public interface Future { 
  2.  
  3.    
  4.     boolean cancel(boolean mayInterruptIfRunning); 
  5.  
  6.     
  7.     boolean isCancelled(); 
  8.  
  9.     
  10.     boolean isDone(); 
  11.  
  12.     
  13.     V get() throws InterruptedException, ExecutionException; 
  14.  
  15.     
  16.     V get(long timeout, TimeUnit unit) 
  17.         throws InterruptedException, ExecutionException, TimeoutException; 

 get方法(获取结果)

get 方法最主要的作用就是获取任务执行的结果,该方法在执行时的行为取决于 Callable 任务的状态,可能会发生以下 7 种情况。

参考示例:

  1. package com.niuh.future; 
  2.  
  3. import java.util.Random; 
  4. import java.util.concurrent.Callable; 
  5. import java.util.concurrent.ExecutionException; 
  6. import java.util.concurrent.ExecutorService; 
  7. import java.util.concurrent.Executors; 
  8. import java.util.concurrent.Future; 
  9. import java.util.concurrent.TimeUnit; 
  10. import java.util.concurrent.TimeoutException; 
  11.  
  12. public class FutureDemo { 
  13.     public static void main(String[] args) { 
  14.         ExecutorService executorService = Executors.newSingleThreadExecutor(); 
  15.         Future<Integer> future = executorService.submit(new FutureTask()); 
  16.         try { 
  17.             Integer res = future.get(2000, TimeUnit.MILLISECONDS); 
  18.             System.out.println("Future线程返回值:" + res); 
  19.         } catch (InterruptedException e) { 
  20.             e.printStackTrace(); 
  21.         } catch (ExecutionException e) { 
  22.             e.printStackTrace(); 
  23.         } catch (TimeoutException e) { 
  24.             e.printStackTrace(); 
  25.         } 
  26.     } 
  27.  
  28.     static class FutureTask implements Callable<Integer> { 
  29.  
  30.         @Override 
  31.         public Integer call() throws Exception { 
  32.             Thread.sleep(new Random().nextInt(3000)); 
  33.             return new Random().nextInt(10); 
  34.         } 
  35.     } 

 isDone方法(判断是否执行完毕)

isDone() 方法,该方法是用来判断当前这个任务是否执行完毕了

  1. package com.niuh.future; 
  2.  
  3. import java.util.Random; 
  4. import java.util.concurrent.Callable; 
  5. import java.util.concurrent.ExecutionException; 
  6. import java.util.concurrent.ExecutorService; 
  7. import java.util.concurrent.Executors; 
  8. import java.util.concurrent.Future; 
  9. import java.util.concurrent.TimeUnit; 
  10. import java.util.concurrent.TimeoutException; 
  11.  
  12. public class FutureIsDoneDemo { 
  13.     public static void main(String[] args) { 
  14.         ExecutorService executorService = Executors.newSingleThreadExecutor(); 
  15.         Future<Integer> future = executorService.submit(new FutureTask()); 
  16.         try { 
  17.             for (int i = 0; i < 3; i++) { 
  18.                 Thread.sleep(1000); 
  19.                 System.out.println("线程是否完成:" + future.isDone()); 
  20.             } 
  21.             Integer res = future.get(2000, TimeUnit.MILLISECONDS); 
  22.             System.out.println("Future 线程返回值:" + res); 
  23.         } catch (InterruptedException e) { 
  24.             e.printStackTrace(); 
  25.         } catch (ExecutionException e) { 
  26.             e.printStackTrace(); 
  27.         } catch (TimeoutException e) { 
  28.             e.printStackTrace(); 
  29.         } 
  30.     } 
  31.  
  32.     static class FutureTask implements Callable<Integer> { 
  33.  
  34.         @Override 
  35.         public Integer call() throws Exception { 
  36.             Thread.sleep(2000); 
  37.             return new Random().nextInt(10); 
  38.         } 
  39.     } 

 执行结果:

  1. 线程是否完成:false 
  2. 线程是否完成:false 
  3. 线程是否完成:true 
  4. Future 线程返回值:9 

 可以看到前两次 isDone 方法的返回结果是 false,因为线程任务还没有执行完成,第三次 isDone 方法的返回结果是 ture。

注意:这个方法返回 true 则代表执行完成了,返回 false 则代表还没完成。但返回 true,并不代表这个任务是成功执行的,比如说任务执行到一半抛出了异常。那么在这种情况下,对于这个 isDone 方法而言,它其实也是会返回 true 的,因为对它来说,虽然有异常发生了,但是这个任务在未来也不会再被执行,它确实已经执行完毕了。所以 isDone 方法在返回 true 的时候,不代表这个任务是成功执行的,只代表它执行完毕了。

我们将上面的示例稍作修改再来看下结果,修改 FutureTask 代码如下:

  1. static class FutureTask implements Callable<Integer> { 
  2.  @Override 
  3.  public Integer call() throws Exception { 
  4.   Thread.sleep(2000); 
  5.   throw new Exception("故意抛出异常"); 
  6.     } 

 执行结果:

虽然抛出了异常,但是 isDone 方法的返回结果依然是 ture。

这段代码说明了:

cancel方法(取消任务的执行)

如果不想执行某个任务了,则可以使用 cancel 方法,会有以下三种情况:

参考示例:

  1. package com.niuh.future; 
  2.  
  3. import java.util.concurrent.Callable; 
  4. import java.util.concurrent.ExecutionException; 
  5. import java.util.concurrent.ExecutorService; 
  6. import java.util.concurrent.Executors; 
  7. import java.util.concurrent.Future; 
  8. import java.util.concurrent.TimeUnit; 
  9. import java.util.concurrent.TimeoutException; 
  10.  
  11. public class FutureCancelDemo { 
  12.  
  13.     static ExecutorService executorService = Executors.newSingleThreadExecutor(); 
  14.  
  15.     public static void main(String[] args) { 
  16.         // 当任务还没有开始执行 
  17.         // demo1(); 
  18.  
  19.         // 如果任务已经执行完 
  20.         // demo2(); 
  21.  
  22.         // 如果任务正在进行中 
  23.         demo3(); 
  24.     } 
  25.  
  26.     private static void demo1() { 
  27.         for (int i = 0; i < 1000; i++) { 
  28.             executorService.submit(new FutureTask()); 
  29.         } 
  30.  
  31.         Future future = executorService.submit(new FutureTask()); 
  32.         try { 
  33.             boolean cancel = future.cancel(false); 
  34.             System.out.println("Future 任务是否被取消:" + cancel); 
  35.             String res = future.get(2000, TimeUnit.MILLISECONDS); 
  36.             System.out.println("Future 线程返回值:" + res); 
  37.         } catch (InterruptedException e) { 
  38.             e.printStackTrace(); 
  39.         } catch (ExecutionException e) { 
  40.             e.printStackTrace(); 
  41.         } catch (TimeoutException e) { 
  42.             e.printStackTrace(); 
  43.         } finally { 
  44.             executorService.shutdown(); 
  45.         } 
  46.     } 
  47.  
  48.  
  49.     private static void demo2() { 
  50.         Future future = executorService.submit(new FutureTask()); 
  51.         try { 
  52.             Thread.sleep(1000); 
  53.             boolean cancel = future.cancel(false); 
  54.             System.out.println("Future 任务是否被取消:" + cancel); 
  55.         } catch (InterruptedException e) { 
  56.             e.printStackTrace(); 
  57.         } finally { 
  58.             executorService.shutdown(); 
  59.         } 
  60.     } 
  61.  
  62.     private static void demo3() { 
  63.         Future future = executorService.submit(new FutureInterruptTask()); 
  64.         try { 
  65.             Thread.sleep(1000); 
  66.             boolean cancel = future.cancel(true); 
  67.             System.out.println("Future 任务是否被取消:" + cancel); 
  68.         } catch (InterruptedException e) { 
  69.             e.printStackTrace(); 
  70.         } finally { 
  71.             executorService.shutdown(); 
  72.         } 
  73.     } 
  74.  
  75.  
  76.     static class FutureTask implements Callable { 
  77.  
  78.         @Override 
  79.         public String call() throws Exception { 
  80.             return "正常返回"
  81.         } 
  82.     } 
  83.  
  84.     static class FutureInterruptTask implements Callable { 
  85.  
  86.         @Override 
  87.         public String call() throws Exception { 
  88.             while (!Thread.currentThread().isInterrupted()) { 
  89.                 System.out.println("循环执行"); 
  90.                 Thread.sleep(500); 
  91.             } 
  92.             System.out.println("线程被中断"); 
  93.             return "正常返回"
  94.         } 
  95.     } 

 这里,我们来分析下第三种情况(任务正在进行中),当我们设置 true 时,线程停止

  1. 循环执行 
  2. 循环执行 
  3. Future 任务是否被取消:true 

 当我们设置 false 时,任务虽然也被取消成功,但是线程依然执行。 

  1. 循环执行 
  2. 循环执行 
  3. Future 任务是否被取消:true 
  4. 循环执行 
  5. 循环执行 
  6. 循环执行 
  7. 循环执行 
  8. ...... 

 那么如何选择传入 true 还是 false 呢?

需要注意的是,虽然示例中写了 !Thread.currentThread().isInterrupted() 方法来判断中断,但是实际上并不是通过我们的代码来进行中断,而是 Future#cancel(true) 内部调用 t.interrupt 方法修改线程的状态之后,Thread.sleep 会抛出 InterruptedException 异常,线程池中会执行异常的相关逻辑,并退出当前任务。 sleep 和 interrupt 会产生意想不到的效果。

比如我们将 FutureInterruptTask 代码修改为 while(true) 形式,调用 cancel(true) 方法线程还是会被中断。 

  1. static class FutureInterruptTask implements Callable { 
  2.  @Override 
  3.  public String call() throws Exception { 
  4.   while (true) { 
  5.             System.out.println("循环执行"); 
  6.             Thread.sleep(500); 
  7.   } 
  8.  } 

 isCancelled方法(判断是否被取消)

isCancelled 方法,判断是否被取消,它和 cancel 方法配合使用,比较简单,可以参考上面的示例。

Callable 和 Future 的关系

Callable 接口相比于 Runnable 的一大优势是可以有返回结果,返回结果就可以用 Future 类的 get 方法来获取 。因此,Future 相当于一个存储器,它存储了 Callable 的 call 方法的任务结果。

除此之外,我们还可以通过 Future 的 isDone 方法来判断任务是否已经执行完毕了,还可以通过 cancel 方法取消这个任务,或限时获取任务的结果等,总之 Future 的功能比较丰富。

FutureTask

Future只是一个接口,不能直接用来创建对象,其实现类是FutureTask,JDK1.8修改了FutureTask的实现,JKD1.8不再依赖AQS来实现,而是通过一个volatile变量state以及CAS操作来实现。FutureTask结构如下所示:

我们来看一下 FutureTask 的代码实现:

  1. public class FutureTask implements RunnableFuture {...} 

可以看到,它实现了一个接口,这个接口叫作 RunnableFuture。

RunnableFuture接口

我们来看一下 RunnableFuture 接口的代码实现:

  1. public interface RunnableFuture extends Runnable, Future { 
  2.      
  3.     void run(); 

 既然 RunnableFuture 继承了 Runnable 接口和 Future 接口,而 FutureTask 又实现了 RunnableFuture 接口,所以 FutureTask 既可以作为 Runnable 被线程执行,又可以作为 Future 得到 Callable 的返回值。

FutureTask源码分析

成员变量

  1.  
  2. private volatile int state; 
  3. private static final int NEW          = 0; // 新建 0 
  4. private static final int COMPLETING   = 1; // 执行中 1 
  5. private static final int NORMAL       = 2; // 正常 2 
  6. private static final int EXCEPTIONAL  = 3; // 异常 3 
  7. private static final int CANCELLED    = 4; // 取消 4 
  8. private static final int INTERRUPTING = 5; // 中断中 5 
  9. private static final int INTERRUPTED  = 6; // 被中断 6 
  10.  
  11.  
  12. private Callable callable; 
  13.  
  14. private Object outcome; // non-volatile, protected by state reads/writes 
  15.  
  16. private volatile Thread runner; 
  17.  
  18. private volatile WaitNode waiters; 

 为了后面更好的分析FutureTask的实现,这里有必要解释下各个状态。

有一点需要注意的是,所有值大于COMPLETING的状态都表示任务已经执行完成(任务正常执行完成,任务执行异常或者任务被取消)。

构造方法

  1. // Callable 构造方法 
  2. public FutureTask(Callable callable) { 
  3.     if (callable == null
  4.         throw new NullPointerException(); 
  5.     this.callable = callable; 
  6.     this.state = NEW;       // ensure visibility of callable 
  7.  
  8. // Runnable 构造方法 
  9. public FutureTask(Runnable runnable, V result) { 
  10.     this.callable = Executors.callable(runnable, result); 
  11.     this.state = NEW;       // ensure visibility of callable 

 Runnable的构造器,只有一个目的,就是通过Executors.callable把入参转化为RunnableAdapter,主要是因为Callable的功能比Runnable丰富,Callable有返回值,而Runnable没有。

  1.  
  2. static final class RunnableAdapter implements Callable { 
  3.     final Runnable task; 
  4.     final T result; 
  5.     RunnableAdapter(Runnable task, T result) { 
  6.         this.task = task; 
  7.         this.result = result; 
  8.     } 
  9.     public T call() { 
  10.         task.run(); 
  11.         return result; 
  12.     } 

 这是一个典型的适配模型,我们要把 Runnable 适配成 Callable,首先要实现 Callable 的接口,接着在 Callable 的 call 方法里面调用被适配对象(Runnable)的方法。

内部类 

  1. static final class WaitNode { 
  2.  volatile Thread thread; 
  3.  volatile WaitNode next
  4.  WaitNode() { thread = Thread.currentThread(); } 

 run方法

  1.  
  2. public void run() { 
  3.  // 状态不是任务创建,或者当前任务已经有线程在执行了,直接返回 
  4.     if (state != NEW || 
  5.         !U.compareAndSwapObject(this, RUNNER, null, Thread.currentThread())) 
  6.         return
  7.     try { 
  8.         Callable c = callable; 
  9.         // Callable 不为空,并且已经初始化完成 
  10.         if (c != null && state == NEW) { 
  11.             V result; 
  12.             boolean ran; 
  13.             try { 
  14.              //调用执行 
  15.                 result = c.call(); 
  16.                 ran = true
  17.             } catch (Throwable ex) { 
  18.                 result = null
  19.                 ran = false;//执行失败 
  20.                 //通过CAS算法设置返回值(COMPLETING)和状态值(EXCEPTIONAL) 
  21.                 setException(ex); 
  22.             } 
  23.             //执行成功通过CAS(UNSAFE)设置返回值(COMPLETING)和状态值(NORMAL) 
  24.             if (ran) 
  25.              //将result赋值给outcome 
  26.                 set(result); 
  27.         } 
  28.     } finally { 
  29.         // runner must be non-null until state is settled to 
  30.         // prevent concurrent calls to run() 
  31.         //将任务runner设置为null,避免发生并发调用run()方法 
  32.         runner = null
  33.         // state must be re-read after nulling runner to prevent 
  34.         // leaked interrupts 
  35.         //须重新读取任务状态,避免不可达(泄漏)的中断 
  36.         int s = state; 
  37.         //确保cancle(ture)操作时,运行中的任务能接收到中断指令 
  38.         if (s >= INTERRUPTING) 
  39.             handlePossibleCancellationInterrupt(s); 
  40.     } 
 
  1. run方法是没有返回值的,通过给outcome属性赋值(set(result)),get时就能从outcome属性中拿到返回值。
  2. FutureTask 两种构造器,最终都转化成了 Callable,所以在 run 方法执行的时候,只需要执行 Callable 的 call 方法即可,在执行 c.call()代码时,如果入参是 Runnable 的话, 调用路径为 c.call() -> RunnableAdapter.call() -> Runnable.run(),如果入参是 Callable 的话,直接调用。

setException(Throwable t)方法

  1. //发生异常时,将返回值设置到outcome(=COMPLETING)中,并更新任务状态(EXCEPTIONAL) 
  2. protected void setException(Throwable t) { 
  3.  //调用UNSAFE类封装的CAS算法,设置值 
  4.  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 
  5.      outcome = t; 
  6.     UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state 
  7.     //唤醒因等待返回值而阻塞的线程 
  8.     finishCompletion(); 
  9.     } 

 由于篇幅有限,更多源码解析请查看文章扩展链接

Future的使用

FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask,直接调用其run方法或者放入线程池执行,之后可以在外部通过FutureTask的get方法异步获取执行结果,因此,FutureTask非常适合用于耗时的计算,主线程可以在完成自己的任务后,再去获取结果。另外,FutureTask还可以确保即使调用了多次run方法,它都只会执行一次Runnable或者Callable任务,或者通过cancel取消FutureTask的执行等。

FutureTask执行多任务计算的使用场景

利用FutureTask和ExecutorService,可以用多线程的方式提交计算任务,主线程继续执行其他任务,当主线程需要子线程的计算结果时,在异步获取子线程的执行结果。

  1. //任务正常完成,将返回值设置到outcome(=COMPLETING)中,并更新任务状态(=NORMAL) 
  2. protected void set(V v) { 
  3.  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 
  4.   outcome = v; 
  5.         UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state 
  6.         finishCompletion(); 
  7.     } 

 执行结果:

  1. 生成子线程计算任务: 0 
  2. 生成子线程计算任务: 1 
  3. 生成子线程计算任务: 2 
  4. 生成子线程计算任务: 3 
  5. 生成子线程计算任务: 4 
  6. 生成子线程计算任务: 5 
  7. 生成子线程计算任务: 6 
  8. 生成子线程计算任务: 7 
  9. 生成子线程计算任务: 8 
  10. 生成子线程计算任务: 9 
  11. 所有计算任务提交完毕, 主线程接着干其他事情! 
  12. 子线程计算任务: 0 执行完成! 
  13. 子线程计算任务: 1 执行完成! 
  14. 子线程计算任务: 3 执行完成! 
  15. 子线程计算任务: 4 执行完成! 
  16. 子线程计算任务: 2 执行完成! 
  17. 子线程计算任务: 5 执行完成! 
  18. 子线程计算任务: 7 执行完成! 
  19. 子线程计算任务: 9 执行完成! 
  20. 子线程计算任务: 8 执行完成! 
  21. 子线程计算任务: 6 执行完成! 
  22. 多任务计算后的总结果是:990 

 FutureTask在高并发环境下确保任务只执行一次

在很多高并发的环境下,往往我们只需要某些任务只执行一次。这种使用情景FutureTask的特性恰能胜任。举一个例子,假设有一个带key的连接池,当key存在时,即直接返回key对应的对象;当key不存在时,则创建连接。对于这样的应用场景,通常采用的方法为使用一个Map对象来存储key和连接池对应的对应关系,典型的代码如下面所示:

  1. //移除所有等待线程并发出信号,调用done(),以及将任务callable清空 
  2. private void finishCompletion() { 
  3.     // assert state > COMPLETING; 
  4.     for (WaitNode q; (q = waiters) != null;) { 
  5.         if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) { 
  6.             //循环唤醒阻塞线程,直到阻塞队列为空 
  7.             for (;;) { 
  8.                 Thread t = q.thread; 
  9.                 if (t != null) { 
  10.                     q.thread = null
  11.                     LockSupport.unpark(t); 
  12.                 } 
  13.                 WaitNode next = q.next
  14.                 //一直到阻塞队列为空,跳出循环 
  15.                 if (next == null
  16.                     break; 
  17.                 q.next = null; // unlink to help gc   方便gc在适当的时候回收 
  18.                 q = next
  19.             } 
  20.             break; 
  21.         } 
  22.     } 
  23.  
  24.     done(); 
  25.  
  26.     callable = null;        // to reduce footprint 

 在上面的例子中,我们通过加锁确保高并发环境下的线程安全,也确保了connection只创建一次,然而却牺牲了性能。改用ConcurrentHash的情况下,几乎可以避免加锁的操作,性能大大提高。 

  1. package com.niuh.future; 
  2.  
  3. import java.sql.Connection
  4. import java.sql.DriverManager; 
  5. import java.sql.SQLException; 
  6. import java.util.concurrent.ConcurrentHashMap; 
  7.  
  8.  
  9. public class FutureTaskConnection2 { 
  10.     private static ConcurrentHashMapConnection> connectionPool = new ConcurrentHashMap<>(); 
  11.  
  12.     public static Connection getConnection(String key) { 
  13.         Connection connection = connectionPool.get(key); 
  14.         if (connection == null) { 
  15.             connection = createConnection(); 
  16.             //根据putIfAbsent的返回值判断是否有线程抢先插入了 
  17.             Connection returnConnection = connectionPool.putIfAbsent(keyconnection); 
  18.             if (returnConnection != null) { 
  19.                 connection = returnConnection; 
  20.             } 
  21.         } else { 
  22.             return connection
  23.         } 
  24.         return connection
  25.     } 
  26.  
  27.     private static Connection createConnection() { 
  28.         try { 
  29.             return DriverManager.getConnection(""); 
  30.         } catch (SQLException e) { 
  31.             e.printStackTrace(); 
  32.         } 
  33.         return null
  34.     } 
  35.  

 但是在高并发的情况下有可能出现Connection被创建多次的现象。 为什么呢?

因为创建Connection是一个耗时操作,假设多个线程涌入getConnection方法,都发现key对应的键不存在,于是所有涌入的线程都开始执行conn=createConnection(),只不过最终只有一个线程能将connection插入到map里。但是这样以来,其它线程创建的的connection就没啥价值,浪费系统开销。

这时最需要解决的问题就是当key不存在时,创建Connection的动作(conn=createConnection();)能放在connectionPool.putIfAbsent()之后执行,这正是FutureTask发挥作用的时机,基于ConcurrentHashMap和FutureTask的改造代码如下:

  1. package com.niuh.future; 
  2.  
  3. import java.sql.Connection
  4. import java.sql.DriverManager; 
  5. import java.sql.SQLException; 
  6. import java.util.concurrent.Callable; 
  7. import java.util.concurrent.ConcurrentHashMap; 
  8. import java.util.concurrent.ExecutionException; 
  9. import java.util.concurrent.FutureTask; 
  10.  
  11.  
  12. public class FutureTaskConnection3 { 
  13.     private static ConcurrentHashMapConnection>> connectionPool = new ConcurrentHashMapConnection>>(); 
  14.  
  15.     public static Connection getConnection(String key) { 
  16.         FutureTask<Connection> connectionFutureTask = connectionPool.get(key); 
  17.         try { 
  18.             if (connectionFutureTask != null) { 
  19.                 return connectionFutureTask.get(); 
  20.             } else { 
  21.                 Callable<Connection> callable = new Callable<Connection>() { 
  22.                     @Override 
  23.                     public Connection call() throws Exception { 
  24.                         return createConnection(); 
  25.                     } 
  26.                 }; 
  27.                 FutureTask<Connection> newTask = new FutureTask<>(callable); 
  28.                 FutureTask<Connection> returnFt = connectionPool.putIfAbsent(key, newTask); 
  29.                 if (returnFt == null) { 
  30.                     connectionFutureTask = newTask; 
  31.                     newTask.run(); 
  32.                 } 
  33.                 return connectionFutureTask.get(); 
  34.             } 
  35.         } catch (ExecutionException e) { 
  36.             e.printStackTrace(); 
  37.         } catch (InterruptedException e) { 
  38.             e.printStackTrace(); 
  39.         } 
  40.         return null
  41.     } 
  42.  
  43.     private static Connection createConnection() { 
  44.         try { 
  45.             return DriverManager.getConnection(""); 
  46.         } catch (SQLException e) { 
  47.             e.printStackTrace(); 
  48.         } 
  49.         return null
  50.     } 

 FutureTask任务执行完回调

FutureTask有一个方法 void done()会在每个线程执行完成return结果时回调。 假设现在需要实现每个线程完成任务执行后主动执行后续任务。

  1. private void handlePossibleCancellationInterrupt(int s) { 
  2.     // It is possible for our interrupter to stall before getting a 
  3.     // chance to interrupt us.  Let's spin-wait patiently. 
  4.     //自旋等待cancle(true)结束(中断结束) 
  5.     if (s == INTERRUPTING) 
  6.         while (state == INTERRUPTING) 
  7.              Thread.yield(); // wait out pending interrupt 
  8.  
  9.     // assert state == INTERRUPTED; 
  10.  
  11.     // We want to clear any interrupt we may have received from 
  12.     // cancel(true).  However, it is permissible to use interrupts 
  13.     // as an independent mechanism for a task to communicate with 
  14.     // its caller, and there is no way to clear only the 
  15.     // cancellation interrupt. 
  16.     // 
  17.     // Thread.interrupted(); 

 执行结果: 

  1. 11:01:37.134 [Thread-0] INFO com.niuh.future.FutureTaskDemo1 - 老板给我来一个月饼 
  2. 11:01:37.139 [pool-1-thread-1] INFO com.niuh.future.FutureTaskDemo1 - 月饼制作中。。。。 
  3. 11:01:37.139 [pool-1-thread-2] INFO com.niuh.future.FutureTaskDemo1 - 月饼制作中。。。。 
  4. 11:01:37.139 [pool-1-thread-3] INFO com.niuh.future.FutureTaskDemo1 - 月饼制作中。。。。 
  5. 11:01:42.151 [pool-1-thread-2] INFO com.niuh.future.FutureTaskDemo1 -  编号[804]月饼已打包好 
  6. 11:01:42.151 [pool-1-thread-3] INFO com.niuh.future.FutureTaskDemo1 -  编号[88]月饼已打包好 
  7. 11:01:42.151 [pool-1-thread-1] INFO com.niuh.future.FutureTaskDemo1 -  编号[166]月饼已打包好 

 PS:以上代码提交在 Github :

https://github.com/Niuh-Study/niuh-juc-final.git

PS:这里有一个技术交流群(扣扣群:1158819530),方便大家一起交流,持续学习,共同进步,有需要的可以加一下。

 

来源:今日头条内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯