Java多線程之 FutureTask:帶有返回值的函數(shù)定義和調(diào)用方式

更新時(shí)間：2021年07月14日 11:39:53 作者：郝偉博士

這篇文章主要介紹了Java多線程之 FutureTask:帶有返回值的函數(shù)定義和調(diào)用方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

FutureTask 返回值的函數(shù)定義和調(diào)用

使用Runnable接口定義的任務(wù)是沒有返回值的。很多時(shí)候，我們是有返回值的，為了解決這個(gè)問題，Java提供了Callable接口，可以返回指定類型的值。

但是這個(gè)接口本身是不具備執(zhí)行能力的，所以Java中，還有一個(gè)FutureTask 類用于使用Callable接口定義帶有返回值的任務(wù)。

使用示例

以下代碼演示了定義和調(diào)用的整個(gè)過程。

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
public class FutureTaskDemo {
	public static void test2() throws Execution{
	    // 基于 Lambda 的 Callable 接口，在new FutureTask中的Lambda表達(dá)式即是Callable接口的實(shí)現(xiàn)
		FutureTask<Integer> task = new FutureTask<>(() -> {
			int t = 0;
			for (int i = 0; i < 10; i++)
				t += i;
			return t;
		}); 
		
		// 使用Thread類執(zhí)行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);
	}
}

執(zhí)行后的輸出：

Start calling.

Finish calling.

Result: 45, Time: 13.620 ms.

Java多線程 FutureTask用法及解析

1 FutureTask概念

FutureTask一個(gè)可取消的異步計(jì)算，F(xiàn)utureTask 實(shí)現(xiàn)了Future的基本方法，提空 start cancel 操作，可以查詢計(jì)算是否已經(jīng)完成，并且可以獲取計(jì)算的結(jié)果。

結(jié)果只可以在計(jì)算完成之后獲取，get方法會(huì)阻塞當(dāng)計(jì)算沒有完成的時(shí)候，一旦計(jì)算已經(jīng)完成，那么計(jì)算就不能再次啟動(dòng)或是取消。

一個(gè)FutureTask 可以用來包裝一個(gè) Callable 或是一個(gè)runnable對(duì)象。因?yàn)镕urtureTask實(shí)現(xiàn)了Runnable方法，所以一個(gè) FutureTask可以提交(submit)給一個(gè)Excutor執(zhí)行(excution).

2 FutureTask使用場(chǎng)景

FutureTask可用于異步獲取執(zhí)行結(jié)果或取消執(zhí)行任務(wù)的場(chǎng)景。

通過傳入Runnable或者Callable的任務(wù)給FutureTask，直接調(diào)用其run方法或者放入線程池執(zhí)行，之后可以在外部通過FutureTask的get方法異步獲取執(zhí)行結(jié)果，因此，F(xiàn)utureTask非常適合用于耗時(shí)的計(jì)算，主線程可以在完成自己的任務(wù)后，再去獲取結(jié)果。

另外，F(xiàn)utureTask還可以確保即使調(diào)用了多次run方法，它都只會(huì)執(zhí)行一次Runnable或者Callable任務(wù)，或者通過cancel取消FutureTask的執(zhí)行等。

2.1 FutureTask執(zhí)行多任務(wù)計(jì)算的使用場(chǎng)景

利用FutureTask和ExecutorService，可以用多線程的方式提交計(jì)算任務(wù)，主線程繼續(xù)執(zhí)行其他任務(wù)，當(dāng)主線程需要子線程的計(jì)算結(jié)果時(shí)，在異步獲取子線程的執(zhí)行結(jié)果。

public class FutureTest1 {
	public static void main(String[] args) {
		Task task = new Task();// 新建異步任務(wù)
		FutureTask<Integer> future = new FutureTask<Integer>(task) {
			// 異步任務(wù)執(zhí)行完成，回調(diào)
			@Override
			protected void done() {
				try {
					System.out.println("future.done():" + get());
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				}
			}
		};
		// 創(chuàng)建線程池（使用了預(yù)定義的配置）
		ExecutorService executor = Executors.newCachedThreadPool();
		executor.execute(future);
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		// 可以取消異步任務(wù)
		// future.cancel(true);
		try {
			// 阻塞，等待異步任務(wù)執(zhí)行完畢-獲取異步任務(wù)的返回值
			System.out.println("future.get():" + future.get());
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
	// 異步任務(wù)
	static class Task implements Callable<Integer> {
		// 返回異步任務(wù)的執(zhí)行結(jié)果
		@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在高并發(fā)環(huán)境下確保任務(wù)只執(zhí)行一次

在很多高并發(fā)的環(huán)境下，往往我們只需要某些任務(wù)只執(zhí)行一次。這種使用情景FutureTask的特性恰能勝任。

舉一個(gè)例子，假設(shè)有一個(gè)帶key的連接池，當(dāng)key存在時(shí)，即直接返回key對(duì)應(yīng)的對(duì)象；當(dāng)key不存在時(shí)，則創(chuàng)建連接。

對(duì)于這樣的應(yīng)用場(chǎng)景，通常采用的方法為使用一個(gè)Map對(duì)象來存儲(chǔ)key和連接池對(duì)應(yīng)的對(duì)應(yīng)關(guān)系，典型的代碼如下面所示：

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{  
            //創(chuàng)建 Connection  
            Connection conn = createConnection();  
            connectionPool.put(key, conn);  
            return conn;  
        }  
    }  
    finally{  
        lock.unlock();  
    }  
}  
  
//創(chuàng)建Connection（根據(jù)業(yè)務(wù)需求，自定義Connection）  
private Connection createConnection(){  
    return null;  
}

在上面的例子中，我們通過加鎖確保高并發(fā)環(huán)境下的線程安全，也確保了connection只創(chuàng)建一次，然而確犧牲了性能。改用ConcurrentHash的情況下，幾乎可以避免加鎖的操作，性能大大提高，但是在高并發(fā)的情況下有可能出現(xiàn)Connection被創(chuàng)建多次的現(xiàn)象。

這時(shí)最需要解決的問題就是當(dāng)key不存在時(shí)，創(chuàng)建Connection的動(dòng)作能放在connectionPool之后執(zhí)行，這正是FutureTask發(fā)揮作用的時(shí)機(jī)，基于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();  
    }  
}  
  
//創(chuàng)建Connection（根據(jù)業(yè)務(wù)需求，自定義Connection）  
private Connection createConnection(){  
    return null;  
}

經(jīng)過這樣的改造，可以避免由于并發(fā)帶來的多次創(chuàng)建連接及鎖的出現(xiàn)。

3 部分源碼分析

3.1 構(gòu)造方法

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

3.2 cancel

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

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: