pycharm 做网站,政务网站建设经验做法,公司做网站费用,做品牌设计公司【多线程】线程池Future和FutureTask 【一】Future概述【1】Future的出现原因【2】Future结构图 【二】Future详解【1】Future接口源码【2】Future的5个方法【3】ThreadPoolExecutor提供了三个方法#xff0c;来获取返回值#xff08;1#xff09;submit(Runnable r)#x… 【多线程】线程池Future和FutureTask 【一】Future概述【1】Future的出现原因【2】Future结构图 【二】Future详解【1】Future接口源码【2】Future的5个方法【3】ThreadPoolExecutor提供了三个方法来获取返回值1submit(Runnable r)2submit(Runnable r,T result)3submit(Callable c) 【推荐】 【4】Future使用案例1Future的案例一2FutureTask的案例二3FutureTask的案例三4FutureTask的案例四 【5】总结 【三】其他案例【1】FutureTask获取分布式锁【2】FutureTask监控任务的处理时长是否超时【3】FutureTask监控任务的处理时长是否超时2 【一】Future概述
【1】Future的出现原因
我们在Java线程池ThreadPoolExecutor使用详解中利用execute(Runnable r)方法来异步执行任务但是有一个缺点就是无法执行带有返回值的任务。
【2】Future结构图 FutureTask实现了RunnableFuture接口而RunnableFuture继承了Runnable和Future也就是说FutureTask既是Runnable也是Future。
【二】Future详解
Future表示一个任务的周期并提供了相应的方法来判断是否已经完成或者取消以及获取任务的结果和取消任务。
1Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果该方法会阻塞直到任务返回结果。
2Future是一个接口是无法生成一个实例的所以又有了FutureTask。FutureTask实现了RunnableFuture接口RunnableFuture接口又实现了Runnable接口和Future接口。所以FutureTask既可以被当做Runnable来执行也可以被当做Future来获取Callable的返回结果。
【1】Future接口源码
public interface FutureV {boolean cancel(boolean mayInterruptIfRunning);boolean isCancelled();boolean isDone();V get() throws InterruptedException, ExecutionException;V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}
【2】Future的5个方法 1、cancel()取消一个任务并返回取消结果。参数表示是否中断线程 2、isCancelled()任务是否取消成功 3、isDone()判断当前任务是否执行完毕包括正常执行完毕、执行异常或者任务取消 4、get()获取任务执行结果任务结束之前会阻塞 5、get(long count,TimeUnit)在指定时间内尝试获取执行结果。若超时则抛出超时异常 【3】ThreadPoolExecutor提供了三个方法来获取返回值 1submit(Runnable r)
因为Runnable的run方法是void的没有返回值所以Future.get()是null。
源码
static final class RunnableAdapterT implements CallableT {final Runnable task;final T result;RunnableAdapter(Runnable task, T result) {this.task task;this.result result;}public T call() {task.run();return result;}
}将runnable封装为一个自定义的Callable对象result为null具体的流程可以看最底下的源码。
2submit(Runnable r,T result)
T result就是传入的对象。我们可以通过自定义Runnable来将result传入到自定义的Runnable中在run()方法中对结果写入到result中。其中返回的对象future.get()result。
案例
public class TestThreadExecutor {public static void main(String[] args) throws ExecutionException, InterruptedException {BlockingQueue queue new LinkedBlockingQueue(10);ThreadPoolExecutor executor new ThreadPoolExecutor(4,10,2,TimeUnit.SECONDS,queue);//设置coreThread如果超时也会被取消executor.allowCoreThreadTimeOut(true);int[] arr new int[2];MyRunnable runnable new MyRunnable(arr);Future f executor.submit(runnable,arr);System.out.println(Arrays.toString((int[])f.get()));}
}
-------------------------------------------------------------------------
class MyRunnable implements Runnable {private int[] arr;public MyRunnable(int[] arr){this.arr arr;}Overridepublic void run() {arr[0] 111;}
} 源码
static final class RunnableAdapterT implements CallableT {final Runnable task;final T result;RunnableAdapter(Runnable task, T result) {this.task task;this.result result;}public T call() {task.run();return result;}
}
将runnable封装为一个自定义的Callable对象result为传入的对象具体的流程可以看最底下的源码。
3submit(Callable c) 【推荐】
submit(Runnable r,T result)不太好用需要每次都自定义Runnable和result太麻烦了。
Callable更加好用传入一个Callable对象就行。 Callable和Runnable是大致一样只不过Callable能够返回值。
Callable callable new Callable() {Overridepublic Object call() throws Exception {String name 777;name 6666;return name;}
};Future f executor.submit(callable);
System.out.println(f.get());从上述代码可以看出Callable中call()返回的对象就是future.get()得到的对象。
源码
public T FutureT submit(CallableT task) {if (task null) throw new NullPointerException();RunnableFutureT ftask newTaskFor(task);//利用callable新建了一个FutureTaskexecute(ftask);//会调用ftask.run()方法return ftask;//返回一个FutureTask对象
}protected T RunnableFutureT newTaskFor(CallableT callable) {return new FutureTaskT(callable);
}public FutureTask(CallableV callable) {if (callable null)throw new NullPointerException();this.callable callable;this.state NEW;
}public void run() {if (state ! NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {CallableV c callable;if (c ! null state NEW) {V result;boolean ran;try {//执行callable.call()result c.call();ran true;} catch (Throwable ex) {result null;ran false;setException(ex);}if (ran)//将outcome设置为resultset(result);}} finally {runner null;int s state;if (s INTERRUPTING)handlePossibleCancellationInterrupt(s);}
}protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {//这里这里这里outcome v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}
}
future.get()
public V get() throws InterruptedException, ExecutionException {int s state;if (s COMPLETING)//阻塞s awaitDone(false, 0L);//这里这里这里return report(s);
}private V report(int s) throws ExecutionException {//这里这里这里Object x outcome;if (s NORMAL)//这里这里这里return (V)x;if (s CANCELLED)throw new CancellationException();throw new ExecutionException((Throwable)x);
}
【4】Future使用案例
1Future的案例一
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;public class Main {public static void main(String[] args) {ExecutorService executorService Executors.newCachedThreadPool();ListFutureString resultList new ArrayListFutureString();// 创建10个任务并执行for (int i 0; i 10; i) {// 使用ExecutorService执行Callable类型的任务并将结果保存在future变量中FutureString future executorService.submit(new TaskWithResult(i));// 将任务执行结果存储到List中resultList.add(future);}executorService.shutdown();// 遍历任务的结果for (FutureString fs : resultList) {try {System.out.println(fs.get()); // 打印各个线程任务执行的结果} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {executorService.shutdownNow();e.printStackTrace();return;}}}
}
class TaskWithResult implements CallableString {private int id;public TaskWithResult(int id) {this.id id;}/*** 任务的具体过程一旦任务传给ExecutorService的submit方法则该方法自动在一个线程上执行。** return* throws Exception*/public String call() throws Exception {System.out.println(call()方法被自动调用,干活 Thread.currentThread().getName());if (new Random().nextBoolean())throw new TaskException(Meet error in task. Thread.currentThread().getName());// 一个模拟耗时的操作for (int i 999999999; i 0; i--);return call()方法被自动调用任务的结果是 id Thread.currentThread().getName();}
}class TaskException extends Exception {public TaskException(String message) {super(message);}
}
执行结果
call()方法被自动调用,干活 pool-1-thread-2
call()方法被自动调用,干活 pool-1-thread-6
call()方法被自动调用,干活 pool-1-thread-1
call()方法被自动调用任务的结果是0 pool-1-thread-1
call()方法被自动调用任务的结果是1 pool-1-thread-2
call()方法被自动调用,干活 pool-1-thread-4
call()方法被自动调用,干活 pool-1-thread-9
call()方法被自动调用,干活 pool-1-thread-8
call()方法被自动调用,干活 pool-1-thread-10
call()方法被自动调用,干活 pool-1-thread-5
call()方法被自动调用,干活 pool-1-thread-3
call()方法被自动调用,干活 pool-1-thread-7
call()方法被自动调用任务的结果是2 pool-1-thread-3
call()方法被自动调用任务的结果是3 pool-1-thread-4
java.util.concurrent.ExecutionException: com.company.TaskException: Meet error in task.pool-1-thread-5
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at com.company.Main.main(Main.java:29)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
Caused by: com.company.TaskException: Meet error in task.pool-1-thread-5
at com.company.TaskWithResult.call(Main.java:56)
at com.company.TaskWithResult.call(Main.java:40)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)2FutureTask的案例二
FutureTask是一个具体的实现类ThreadPoolExecutor的submit方法返回的就是一个Future的实现这个实现就是FutureTask的一个具体实例FutureTask帮助实现了具体的任务执行以及和Future接口中的get方法的关联。
import java.util.Date;
import java.text.SimpleDateFormat;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;public class Main {public static void main(String[] args) {SimpleDateFormat df new SimpleDateFormat(yyyy-MM-dd HH:mm:ss);// 初始化一个Callable对象和FutureTask对象Callable pAccount new PrivateAccount();FutureTask futureTask new FutureTask(pAccount);// 使用futureTask创建一个线程Thread pAccountThread new Thread(futureTask);System.out.println(futureTask线程现在开始启动启动时间为 df.format(new Date()));pAccountThread.start();System.out.println(futureTask线程开始执行其他任务);// 从其他账户获取总金额int totalMoney new Random().nextInt(100000);System.out.println(现在你在其他账户中的总金额为 totalMoney);System.out.println(等待私有账户总金额统计完毕...);// 测试后台的计算线程是否完成如果未完成则等待while (!futureTask.isDone()) {try {Thread.sleep(500);System.out.println(私有账户计算未完成继续等待...);} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(futureTask线程计算完毕此时时间为 df.format(new Date()));Integer privateAccountMoney null;try {privateAccountMoney (Integer) futureTask.get();} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}System.out.println(您现在的总金额为 totalMoney ||privateAccountMoney.intValue());}
}class PrivateAccount implements Callable {Integer totalMoney;Overridepublic Object call() throws Exception {Thread.sleep(5000);totalMoney new Integer(new Random().nextInt(10000));System.out.println(您当前有 totalMoney 在您的私有账户中);return totalMoney;}
}
执行结果
futureTask线程现在开始启动启动时间为2016-11-30 16:34:57
futureTask线程开始执行其他任务
现在你在其他账户中的总金额为17113
等待私有账户总金额统计完毕…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
私有账户计算未完成继续等待…
您当前有8776在您的私有账户中
私有账户计算未完成继续等待…
futureTask线程计算完毕此时时间为2016-11-30 16:35:02
您现在的总金额为17113||87763FutureTask的案例三
import java.util.Date;
import java.text.SimpleDateFormat;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;public class Main {public static void main(String[] args) {SimpleDateFormat df new SimpleDateFormat(yyyy-MM-dd HH:mm:ss);FutureTaskString futureTasknew FutureTask(new CallableString() {//Overridepublic String call() throws Exception {// TODO Auto-generated method stubreturn 回调完成;}});Thread thread new Thread(futureTask);System.out.println(启动时间为 df.format(new Date()));thread.start();try {String strfutureTask.get();if(str.equals(回调完成))System.out.println(异步任务完成!);elseSystem.out.println(Completed!);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (ExecutionException e) {// TODO Auto-generated catch blocke.printStackTrace();}}
}
执行结果
启动时间为2016-12-01 09:37:03
异步任务完成!4FutureTask的案例四
public static void main(String[] args) {ExecutorService service Executors.newCachedThreadPool();Task task new Task();FutureTaskInteger futureTask new FutureTask(task);service.submit(futureTask);service.shutdown();System.out.println(主线程正在执行任务。。。);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}try {//阻塞直至任务完成System.out.println(-------------------------------------);System.out.println(执行结果为 futureTask.get());} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}System.out.println(任务执行完成。。。);}static class Task implements CallableInteger {Overridepublic Integer call() throws Exception {try {System.out.println(从线程正在执行任务。。。);Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}int num 0;for (int i 0; i 100; i) {num i;}return num;}}
futureTask.get()执行时如果该任务已经执行完了则直接返回执行结果如果没有执行完则线程会阻塞在这里直至任务执行完毕。还可以用get(long timeout, TimeUnit unit)来获取执行结果如果在指定时间内还没获取到结果就直接返回null。
【5】总结
总的来说Future模式的思想就是在子线程进行执行的时候主线程不阻塞继续执行。等到主线程需要子线程的结果的时候再去获取子线程的结果此时子线程没有执行完成的话就会阻塞直至执行完成。主线程还可以根据一定的业务逻辑去判断是否要取消执行子线程还可以设置一个超时时间。若阻塞时间超过了超时时间子线程仍然没有执行完成的话可以直接返回null。
【三】其他案例
【1】FutureTask获取分布式锁
String taskInstanceKey getTaskInstanceKey(beSubmitedTask);
//对实例加锁
RLock instanceLock redissonClient.getLock(taskInstanceKey);
FutureTaskBoolean futureTask new FutureTask(() - instanceLock.tryLock());
new Thread(futureTask).start();
if (!futureTask.get()) {log.info(指标固化获取锁失败{}, taskInstanceKey);return;
}使用futureTask 取调用tryLock尝试获取锁如果返回值为空说明获取锁失败
【2】FutureTask监控任务的处理时长是否超时
这里采用FutureTask开启一个线程调用jdbc执行方法主方法在10分钟后进行查询是否完成根据isDone的结果去处理简单介绍下FutureTask几个常用的方法 1cancel方法用来取消任务如果取消任务成功则返回true如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务如果设置true则表示可以取消正在执行过程中的任务。如果任务已经完成则无论mayInterruptIfRunning为true还是false此方法肯定返回false即如果取消已经完成的任务会返回false如果任务正在执行若mayInterruptIfRunning设置为true则返回true若mayInterruptIfRunning设置为false则返回false如果任务还没有执行则无论mayInterruptIfRunning为true还是false肯定返回true。 2isCancelled方法表示任务是否被取消成功如果在任务正常完成前被取消成功则返回 true。 3isDone方法表示任务是否已经完成若任务完成则返回true 4get()方法用来获取执行结果这个方法会产生阻塞会一直等到任务执行完毕才返回 5get(long timeout, TimeUnit unit)用来获取执行结果如果在指定时间内还没获取到结果就直接返回null。 这里使用的是isDone去完成当然也可以使用et(long timeout, TimeUnit unit)定时去获取返回结果 上代码
public static void readSqlThred(String sqlTest) {ExecutorService executor Executors.newSingleThreadExecutor();FutureTaskString future new FutureTaskString(new CallableString() {Overridepublic String call() throws Exception {dealWithSql(sqlTest);//执行sqlreturn 执行完成;}});executor.execute(future);try {Thread.sleep(1000*6);if(future.isDone()) {System.out.println(sql执行完毕);}else {System.out.println(警告sql未执行完毕开始发送邮件..............);}}catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}
}
这里我是让线程睡了10分钟10分钟后去取这个方法是否执行完成
【3】FutureTask监控任务的处理时长是否超时2
1程序中某方法挺耗时但想超过XX数值则按失败算 2程序中需要调用外部接口要考虑外部接口不可用或卡死问题。
针对以上问题我们都想在程序中给某方法添加超时判断运行程序等待该方法多长时间超过则另作处理。 public static void main(String[] args){Thread t null;try{FutureTaskBoolean task new FutureTaskBoolean(new CallableBoolean() {public Boolean call() throws Exception {Thread.sleep(10*100);//处理方法return true; //返回处理结果}});t new Thread(task);t.start();boolean bl task.get(10*10, TimeUnit.MILLISECONDS); //允许阻塞等待方法处理1秒超出则抛异常System.out.println(bl);System.out.println(t.getState());}catch(Exception e1){e1.printStackTrace();System.out.println(t.getState());t.interrupt(); //在超时情况下上面新生成的线程不会关闭直到方法处理完毕需要中断}}
