Java多線程之 FutureTask:帶有返回值的函數定義和調用方式

FutureTask 返回值的函數定義和調用

使用Runnable接口定義的任務是沒有返回值的。很多時候,我們是有返回值的,為瞭解決這個問題,Java提供瞭Callable接口,可以返回指定類型的值。

但是這個接口本身是不具備執行能力的,所以Java中,還有一個FutureTask 類用於使用Callable接口定義帶有返回值的任務。

使用示例

以下代碼演示瞭定義和調用的整個過程。

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
public class FutureTaskDemo {
	public static void test2() throws Execution{
	    // 基於 Lambda 的 Callable 接口,在new FutureTask中的Lambda表達式即是Callable接口的實現
		FutureTask<Integer> task = new FutureTask<>(() -> {
			int t = 0;
			for (int i = 0; i < 10; i++)
				t += i;
			return t;
		}); 
		
		// 使用Thread類執行task
		System.out.println("Start calling.");
		long t1 = System.nanoTime();
		new Thread(task).start();
		long result = task.get();
		long t2 = System.nanoTime();
		System.out.println("Finish calling.");
		System.out.printf("Result: %d, Time: %.3f ms.\n", result, (t2 - t1) / 1000000f);
	}
}	

執行後的輸出:

Start calling.

Finish calling.

Result: 45, Time: 13.620 ms.

Java多線程 FutureTask用法及解析

1 FutureTask概念

FutureTask一個可取消的異步計算,FutureTask 實現瞭Future的基本方法,提空 start cancel 操作,可以查詢計算是否已經完成,並且可以獲取計算的結果。

結果隻可以在計算完成之後獲取,get方法會阻塞當計算沒有完成的時候,一旦計算已經完成,那麼計算就不能再次啟動或是取消。

一個FutureTask 可以用來包裝一個 Callable 或是一個runnable對象。因為FurtureTask實現瞭Runnable方法,所以一個 FutureTask可以提交(submit)給一個Excutor執行(excution).

2 FutureTask使用場景

FutureTask可用於異步獲取執行結果或取消執行任務的場景。

通過傳入Runnable或者Callable的任務給FutureTask,直接調用其run方法或者放入線程池執行,之後可以在外部通過FutureTask的get方法異步獲取執行結果,因此,FutureTask非常適合用於耗時的計算,主線程可以在完成自己的任務後,再去獲取結果。

另外,FutureTask還可以確保即使調用瞭多次run方法,它都隻會執行一次Runnable或者Callable任務,或者通過cancel取消FutureTask的執行等。

2.1 FutureTask執行多任務計算的使用場景

利用FutureTask和ExecutorService,可以用多線程的方式提交計算任務,主線程繼續執行其他任務,當主線程需要子線程的計算結果時,在異步獲取子線程的執行結果。

public class FutureTest1 {
	public static void main(String[] args) {
		Task task = new Task();// 新建異步任務
		FutureTask<Integer> future = new FutureTask<Integer>(task) {
			// 異步任務執行完成,回調
			@Override
			protected void done() {
				try {
					System.out.println("future.done():" + get());
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				}
			}
		};
		// 創建線程池(使用瞭預定義的配置)
		ExecutorService executor = Executors.newCachedThreadPool();
		executor.execute(future);
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		// 可以取消異步任務
		// future.cancel(true);
		try {
			// 阻塞,等待異步任務執行完畢-獲取異步任務的返回值
			System.out.println("future.get():" + future.get());
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
	// 異步任務
	static class Task implements Callable<Integer> {
		// 返回異步任務的執行結果
		@Override
		public Integer call() throws Exception {
			int i = 0;
			for (; i < 10; i++) {
				try {
					System.out.println(Thread.currentThread().getName() + "_"
							+ i);
					Thread.sleep(500);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			return i;
		}
	}
}

這裡寫圖片描述

2.2 FutureTask在高並發環境下確保任務隻執行一次

在很多高並發的環境下,往往我們隻需要某些任務隻執行一次。這種使用情景FutureTask的特性恰能勝任。

舉一個例子,假設有一個帶key的連接池,當key存在時,即直接返回key對應的對象;當key不存在時,則創建連接。

對於這樣的應用場景,通常采用的方法為使用一個Map對象來存儲key和連接池對應的對應關系,典型的代碼如下面所示:

private Map<String, Connection> connectionPool = new HashMap<String, Connection>();  
private ReentrantLock lock = new ReentrantLock();    
public Connection getConnection(String key){  
    try{  
        lock.lock();  
        if(connectionPool.containsKey(key)){  
            return connectionPool.get(key);  
        }  
        else{  
            //創建 Connection  
            Connection conn = createConnection();  
            connectionPool.put(key, conn);  
            return conn;  
        }  
    }  
    finally{  
        lock.unlock();  
    }  
}  
  
//創建Connection(根據業務需求,自定義Connection)  
private Connection createConnection(){  
    return null;  
}  

在上面的例子中,我們通過加鎖確保高並發環境下的線程安全,也確保瞭connection隻創建一次,然而確犧牲瞭性能。改用ConcurrentHash的情況下,幾乎可以避免加鎖的操作,性能大大提高,但是在高並發的情況下有可能出現Connection被創建多次的現象。

這時最需要解決的問題就是當key不存在時,創建Connection的動作能放在connectionPool之後執行,這正是FutureTask發揮作用的時機,基於ConcurrentHashMap和FutureTask的改造代碼如下:

private ConcurrentHashMap<String,FutureTask<Connection>>connectionPool = new ConcurrentHashMap<String, FutureTask<Connection>>();  
  
public Connection getConnection(String key) throws Exception{  
    FutureTask<Connection>connectionTask=connectionPool.get(key);  
    if(connectionTask!=null){  
        return connectionTask.get();  
    }  
    else{  
        Callable<Connection> callable = new Callable<Connection>(){  
            @Override  
            public Connection call() throws Exception {  
                // TODO Auto-generated method stub  
                return createConnection();  
            }  
        };  
        FutureTask<Connection>newTask = new FutureTask<Connection>(callable);  
        connectionTask = connectionPool.putIfAbsent(key, newTask);  
        if(connectionTask==null){  
            connectionTask = newTask;  
            connectionTask.run();  
        }  
        return connectionTask.get();  
    }  
}  
  
//創建Connection(根據業務需求,自定義Connection)  
private Connection createConnection(){  
    return null;  
}  

經過這樣的改造,可以避免由於並發帶來的多次創建連接及鎖的出現。

3 部分源碼分析

3.1 構造方法

public FutureTask(Runnable runnable, V result) {  
        this.callable = Executors.callable(runnable, result);  
        this.state = NEW;       // ensure visibility of callable  
 } 

3.2 cancel

//這個方法有一個參數 是否中斷running  
     public boolean cancel(boolean mayInterruptIfRunning) {  
          /** 
          * 這個有點暈啊邏輯關系是 
          * 等價與 if(state!=new || !UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)) 
          * 這個意思是 如果state不是new 那麼就退出方法,這時的任務任務坑是已經完成瞭 或是被取消瞭 或是被中斷瞭 
          * 如果是state 是new 就設置state 為中斷狀態 或是取消狀態 
          * 
          **/  
        if (!(state == NEW &&  
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,  
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))  
            return false;  
        try {    // in case call to interrupt throws exception  
            //如果是可中斷 那麼就 調用系統中斷方法 然後把狀態設置成INTERRUPTED  
            if (mayInterruptIfRunning) {  
                try {  
                    Thread t = runner;  
                    if (t != null)  
                        t.interrupt();  
                } finally { // final state  
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);  
                }  
            }  
        } finally {  
            finishCompletion();  
        }  
        return true;  
    }  

以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。

推薦閱讀: