private void handlePossibleCancellationInterrupt(int s) {
         if (s == INTERRUPTING) // 當(dāng)state為INTERRUPTING時
             while (state == INTERRUPTING) // 表示有線程正在中斷當(dāng)前線程
                 Thread.yield(); // 讓出CPU，自旋等待中斷
     }

再啰嗦下: run方法重點做了以下幾件事：

• 將runner屬性設(shè)置成當(dāng)前正在執(zhí)行run方法的線程

• 調(diào)用callable成員變量的call方法來執(zhí)行任務(wù)

• 設(shè)置執(zhí)行結(jié)果outcome, 如果執(zhí)行成功, 則outcome保存的就是執(zhí)行結(jié)果；如果執(zhí)行過程中發(fā)生了異常, 則outcome中保存的就是異常，設(shè)置結(jié)果之前，先將state狀態(tài)設(shè)為中間態(tài)

• 對outcome的賦值完成后，設(shè)置state狀態(tài)為終止態(tài)(NORMAL或者EXCEPTIONAL)

• 喚醒Treiber棧中所有等待的線程

• 善后清理(waiters, callable，runner設(shè)為null)

• 檢查是否有遺漏的中斷，如果有，等待中斷狀態(tài)完成。

怎么能少了get方法呢,一直阻塞獲取參見:awaitDone

順便偷偷看下get(long, TimeUnit),就是get的方法擴展,增加了超時時間,超時后我還沒拿到就生氣拋異?！?

public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (unit == null) // 參數(shù)校驗
            throw new NullPointerException();
        int s = state; // 執(zhí)行器狀態(tài)
        if (s <= COMPLETING && (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING) // 如果狀態(tài)小于等于COMPLETING，說明任務(wù)正在執(zhí)行，需要等待；等待指定時間，state依然小于等于COMPLETING
            throw new TimeoutException(); // 拋出超時異常
        return report(s); // 報告結(jié)果
    }

那么再看awaitDone,要知道會寫死循環(huán)while(true)|for (;;)的都是高手~

至此,線程安排任務(wù)和獲取我就不啰嗦了~~~~還要很多探索的,畢竟帶薪聊天比較緊張,我就不多贅述了~

隊列

接著我們來看隊列，在FutureTask中，隊列的實現(xiàn)是一個單向鏈表，它表示所有等待任務(wù)執(zhí)行完畢的線程的集合。我們知道，F(xiàn)utureTask實現(xiàn)了Future接口，可以獲取“Task”的執(zhí)行結(jié)果，那么如果獲取結(jié)果時，任務(wù)還沒有執(zhí)行完畢怎么辦呢？那么獲取結(jié)果的線程就會在一個等待隊列中掛起，直到任務(wù)執(zhí)行完畢被喚醒。這一點有點類似于AQS中的sync queue，在下文的分析中，大家可以自己對照它們的異同點。

我們前面說過，在并發(fā)編程中使用隊列通常是將當(dāng)前線程包裝成某種類型的數(shù)據(jù)結(jié)構(gòu)扔到等待隊列中，我們先來看看隊列中的每一個節(jié)點是怎么個結(jié)構(gòu)：

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

可見，相比于AQS的sync queue所使用的雙向鏈表中的Node，這個WaitNode要簡單多了，它只包含了一個記錄線程的thread屬性和指向下一個節(jié)點的next屬性。

值得一提的是，F(xiàn)utureTask中的這個單向鏈表是當(dāng)做棧來使用的，確切來說是當(dāng)做Treiber棧來使用的，不了解Treiber棧是個啥的可以簡單的把它當(dāng)做是一個線程安全的棧，它使用CAS來完成入棧出棧操作(想進一步了解的話可以看這篇文章)。

為啥要使用一個線程安全的棧呢，因為同一時刻可能有多個線程都在獲取任務(wù)的執(zhí)行結(jié)果，如果任務(wù)還在執(zhí)行過程中，則這些線程就要被包裝成WaitNode扔到Treiber棧的棧頂，即完成入棧操作，這樣就有可能出現(xiàn)多個線程同時入棧的情況，因此需要使用CAS操作保證入棧的線程安全，對于出棧的情況也是同理。

由于FutureTask中的隊列本質(zhì)上是一個Treiber(驅(qū)動)棧，那么使用這個隊列就只需要一個指向棧頂節(jié)點的指針就行了，在FutureTask中，就是waiters屬性：

事實上，它就是整個單向鏈表的頭節(jié)點。

綜上，F(xiàn)utureTask中所使用的隊列的結(jié)構(gòu)如下：

CAS操作

CAS操作大多數(shù)是用來改變狀態(tài)的，在FutureTask中也不例外。我們一般在靜態(tài)代碼塊中初始化需要CAS操作的屬性的偏移量：

    // Unsafe mechanics
    private static final sun.misc.Unsafe UNSAFE;
    private static final long stateOffset;
    private static final long runnerOffset;
    private static final long waitersOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = FutureTask.class;
            stateOffset = UNSAFE.objectFieldOffset(k.getDeclaredField('state'));
            runnerOffset = UNSAFE.objectFieldOffset(k.getDeclaredField('runner'));
            waitersOffset = UNSAFE.objectFieldOffset(k.getDeclaredField('waiters'));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

從這個靜態(tài)代碼塊中我們也可以看出，CAS操作主要針對3個屬性，包括state、runner和waiters，說明這3個屬性基本是會被多個線程同時訪問的。其中state屬性代表了任務(wù)的狀態(tài)，waiters屬性代表了指向棧頂節(jié)點的指針，這兩個我們上面已經(jīng)分析過了。

runner屬性代表了執(zhí)行FutureTask中的“Task”的線程。為什么需要一個屬性來記錄執(zhí)行任務(wù)的線程呢？這是為了中斷或者取消任務(wù)做準備的，只有知道了執(zhí)行任務(wù)的線程是誰，我們才能去中斷它。

定義完屬性的偏移量之后，接下來就是CAS操作本身了。在FutureTask，CAS操作最終調(diào)用的還是Unsafe類的compareAndSwapXXX方法，關(guān)于Unsafe，由于帶薪碼文這里不再贅述。

實戰(zhàn)演練

一切沒有例子的講解都是耍流氓 >>> 蔥姜切沫~~加入生命的源泉….

實戰(zhàn)項目以springboot為項目腳手架,github地址:

https://github.com/leaJone/mybot

1.MyFutureTask實現(xiàn)類

內(nèi)部定義一個線程池進行任務(wù)的調(diào)度和線程的管理以及線程的復(fù)用,大家可以根據(jù)自己的實際項目情況進行配置

其中線程調(diào)度示例:核心線程 8 最大線程 20 保活時間30s 存儲隊列 10 有守護線程 拒絕策略:將超負荷任務(wù)回退到調(diào)用者

說明 :

默認使用核心線程(8)數(shù)執(zhí)行任務(wù),任務(wù)數(shù)量超過核心線程數(shù)就丟到隊列,隊列(10)滿了就再開啟新的線程,新的線程數(shù)最大為20,當(dāng)任務(wù)執(zhí)行完,新開啟的線程將存活30s,若沒有任務(wù)就消亡,線程池回到核心線程數(shù)量.

2.service業(yè)務(wù)方法

常規(guī)業(yè)務(wù)查詢方法,為了特效,以及看出實際的效果,我們每個方法做了延時

import com.boot.lea.mybot.mapper.UserMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

@Service
public class UserServiceImpl implements UserService {

    @Autowired
    UserMapper userMapper;

    @Override
    public long countFansCountByUserId(Long userId) {
        try {
            Thread.sleep(10000);
            System.out.println('獲取FansCount===睡眠:' + 10 + 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println('UserService獲取FansCount的線程  ' + Thread.currentThread().getName());
        return 520;
    }

    @Override
    public long countMsgCountByUserId(Long userId) {
        System.out.println('UserService獲取MsgCount的線程  ' + Thread.currentThread().getName());
        try {
            Thread.sleep(10000);
            System.out.println('獲取MsgCount===睡眠:' + 10 + 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 618;
    }

    @Override
    public long countCollectCountByUserId(Long userId) {
        System.out.println('UserService獲取CollectCount的線程  ' + Thread.currentThread().getName());
        try {
            Thread.sleep(10000);
            System.out.println('獲取CollectCount==睡眠:' + 10 + 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 6664;
    }

    @Override
    public long countFollowCountByUserId(Long userId) {
        System.out.println('UserService獲取FollowCount的線程  ' + Thread.currentThread().getName());
        try {
            Thread.sleep(10000);
            System.out.println('獲取FollowCount===睡眠:' + 10+ 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return userMapper.countFollowCountByUserId(userId);
    }

    @Override
    public long countRedBagCountByUserId(Long userId) {
        System.out.println('UserService獲取RedBagCount的線程  ' + Thread.currentThread().getName());
        try {
            TimeUnit.SECONDS.sleep(4);
            System.out.println('獲取RedBagCount===睡眠:' + 4 + 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 99;
    }

    @Override
    public long countCouponCountByUserId(Long userId) {
        System.out.println('UserService獲取CouponCount的線程  ' + Thread.currentThread().getName());
        try {
            TimeUnit.SECONDS.sleep(8);
            System.out.println('獲取CouponCount===睡眠:' + 8+ 's');
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 66;
    }
}

3.controller調(diào)用

我們啟動項目:開啟調(diào)用 http://localhost:8080/api/user/get/data?userId=4

當(dāng)我們線程池配置為:核心線程 8 最大線程 20 ?；顣r間30s 存儲隊列 10 的時候,我們測試的結(jié)果如下:

結(jié)果:我們看到每個server method的執(zhí)行線程都是從線程池中發(fā)起的線程名:User_Async_FutureTask-%d, 總耗時從累計的52秒縮短到10秒,即取決于最耗時的方法查詢時間.

那我們再將注釋代碼放開,進行串行查詢進行測試:

結(jié)果:我們使用串行的方式進行查詢,結(jié)果匯總將達到52秒,那太可怕了~~

總結(jié)

使用FutureTask的時候,就是將任務(wù)runner以caller的方式進行回調(diào),阻塞獲取,最后我們將結(jié)果匯總,即完成了開啟多線程異步調(diào)用我們的業(yè)務(wù)方法.

            Future<Long> fansCountFT = executor.submit(new Callable<Long>() {
                @Override
                public Long call() throws Exception {
                    return userService.countFansCountByUserId(userId);
                }
            });

這里使用的只是一個簡單的例子,具體項目可以定義具體的業(yè)務(wù)方法進行歸并處理,其實在JDK1.8以后,又有了ExecutorCompletionService,ForkJoinTask,CompletableFuture這些都可以實現(xiàn)上述的方法,我們后續(xù)會做一些這些方法使用的案例,期望大家的關(guān)注,文章中有不足之處,歡迎指正~

小甜點

所以:我們要用到親愛的Spring的異步編程,異步編程有很多種方式:比如常見的Future的sync,CompletableFuture.supplyAsync,@Async,哈哈 其實都離不開Thread.start()…,等等我說個笑話:

老爸有倆孩子：小紅和小明。老爸想喝酒了，他讓小紅去買酒，小紅出去了。然后老爸突然想吸煙了，于是老爸讓小明去買煙。在面對對象的思想中，一般會把買東西，然后買回來這件事作為一個方法，如果按照順序結(jié)構(gòu)或者使用多線程同步的話，小明想去買煙就必須等小紅這個買東西的操作進行完。這樣無疑增加了時間的開銷(萬一老爸尿憋呢?)。異步就是為了解決這樣的問題。你可以分別給小紅小明下達指令，讓他們?nèi)ベI東西，然后你就可以自己做自己的事，等他們買回來的時候接收結(jié)果就可以了。

運行結(jié)果: