自定义FutureTask实现

我不是女神ヾ 2022-09-03 00:09 174阅读 0赞

## FutureTask ##

FutureTask是Future的实现，用来异步任务的获取结果，可以启动和取消异步任务，查询异步任务是否计算结束以及获取最终的异步任务的结果。通过get()方法来获取异步任务的结果，但是会阻塞当前线程直至异步任务执行结束。一旦任务执行结束，任务不能重新启动或取消，除非调用runAndReset()方法

代码示例：

public class ThreadTest_Demo { 
        static ExecutorService executors = Executors.newScheduledThreadPool(2);
    
        public ThreadTest_Demo() { 
        }
    
        public static void main(String[] args) throws ExecutionException, InterruptedException { 
            Callable<String> callable = new Callable<String>() { 
                public String call() throws Exception { 
                    String str = "返回某个数据!";
                    return str;
                }
            };
            Future<String> submit = executors.submit(callable);
            KaneFutureTask<String> futureTask = new KaneFutureTask(callable);
            (new Thread(futureTask)).start();
            System.out.println((String)futureTask.get());
        }
    }

实现一个自己的FutureTask

根据FutureTask核心原理，要实现一个FutureTask必须满足以下方面：

*  需要泛型定义用以返回结果类型
 *  需要一个callable对象，在构造方法中传入
 *  需要实现runnable接口，在run方法中实现具体结果计算
 *  需要一个公开的get方法来获取结果
 *  如果线程没有执行完，则调用get方法的线程需要进入等待队列
 *  需要一个字段记录线程执行的状态
 *  需要一个等待队列存储等待结果的线程

代码示例：

public class KaneFutureTask<T> implements Runnable { 
        private Callable<T> callable;
        T result;//返回的结果
        volatile String state = "NEW";//线程执行的状态
        LinkedBlockingQueue<Thread> waiters = new LinkedBlockingQueue();//存储等待结果的线程
    
        public KaneFutureTask(Callable<T> callable) { 
            this.callable = callable;
        }
    
    	//run()方法未执行未必,阻塞
        public T get() { 
            if ("END".equals(this.state)) { 
                return this.result;
            } else { 
                while(!"END".equals(this.state)) { //这里开始阻塞
                	//把线程存放到容器中
                    this.waiters.offer(Thread.currentThread());
                    //阻塞
                    LockSupport.park();
                }
    
                return this.result;
            }
        }
    
        public void run() { 
            try { 
            	//执行Callable的call方法
                this.result = this.callable.call();
            } catch (Exception var5) { 
                var5.printStackTrace();
            } finally { 
            	//线程状态修改
                this.state = "END";
            }
            
    		//从容器中取出当前线程
    		Thread waiter = (Thread)this.waiters.poll();
            for(waiter != null) { 
            	//取消阻塞
                LockSupport.unpark(waiter);
                waiter = (Thread)this.waiters.poll()
            }
    
        }
    }