RxJava2 實戰知識梳理(14) 在 token 過時時，刷新過時 token 並從新發起請求

1、應用背景

首先要感謝簡友 楠柯壹夢 提供的實戰案例，這篇文章的例子是基於他提出的須要在token失效時，刷新token並從新請求接口的應用場景所想到的解決方案。若是你們有別的案例或者在實際中遇到什麼問題也能夠私信我，讓咱們一塊兒完善這系列的文章。

有時候，咱們的某些接口會依賴於用戶的token信息，像咱們項目當中的資訊評論列表、或者帳戶的書籤同步都會依賴於用戶token信息，可是token每每會有必定的有效期，那麼咱們在請求這些接口返回token失效的時候，就須要刷新token再從新發起一次請求，這個流程圖能夠概括以下：

這個應用的場景和 RxJava2 實戰知識梳理(6) - 基於錯誤類型的重試請求 中介紹的場景很相似，以前提到的錯誤類型就指的是 token失效，可是相比以前的例子，咱們增長了額外的兩個需求：

• 在重試以前，須要先去刷新一次token，而不是單純地等待一段時間再重試。
• 若是有多個請求都出現了因token失效而須要從新刷新token的狀況，那麼須要判斷當前是否有另外一個請求正在刷新token，若是有，那麼就不要發起刷新token的請求，而是等待刷新token的請求返回後，直接進行重試。

本文的代碼能夠經過 RxSample 的第十四章獲取。

2、示例講解

2.1 Token 存儲模塊

首先，咱們須要一個地方來緩存須要的Token，這裏用SharedPreferences來實現，有想了解其內部實現原理的同窗能夠看這篇文章：Android 數據存儲知識梳理(3) - SharedPreference 源碼解析。

public class Store {

    private static final String SP_RX = "sp_rx";
    private static final String TOKEN = "token";

    private SharedPreferences mStore;

    private Store() {
        mStore = Utils.getAppContext().getSharedPreferences(SP_RX, Context.MODE_PRIVATE);
    }

    public static Store getInstance() {
        return Holder.INSTANCE;
    }

    private static final class Holder {
        private static final Store INSTANCE = new Store();
    }

    public void setToken(String token) {
        mStore.edit().putString(TOKEN, token).apply();
    }

    public String getToken() {
        return mStore.getString(TOKEN, "");
    }
}
複製代碼

2.2 依賴於 token 的接口

這裏，咱們用一個簡單的getUserObservable來模擬依賴於token的接口，token存儲的是獲取的時間，爲了演示方便，咱們設置若是距離上次獲取的時間大於2s，那麼就認爲過時，並拋出token失效的錯誤，不然調用onNext方法返回接口給下游。

private Observable<String> getUserObservable (final int index, final String token) {
        return Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(TAG, index + "使用token=" + token + "發起請求");
                //模擬根據Token去請求信息的過程。
                if (!TextUtils.isEmpty(token) && System.currentTimeMillis() - Long.valueOf(token) < 2000) {
                    e.onNext(index + ":" + token + "的用戶信息");
                } else {
                    e.onError(new Throwable(ERROR_TOKEN));
                }
            }
        });
    }
複製代碼

2.3 完整的請求過程

下面，咱們來看一下整個完整的請求過程：

private void startRequest(final int index) {
        Observable<String> observable = Observable.defer(new Callable<ObservableSource<String>>() {
            @Override
            public ObservableSource<String> call() throws Exception {
                String cacheToken = TokenLoader.getInstance().getCacheToken();
                Log.d(TAG, index + "獲取到緩存Token=" + cacheToken);
                return Observable.just(cacheToken);
            }
        }).flatMap(new Function<String, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(String token) throws Exception {
                return getUserObservable(index, token);
            }
        }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {

            private int mRetryCount = 0;

            @Override
            public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {

                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Exception {
                        Log.d(TAG, index + ":" + "發生錯誤=" + throwable + ",重試次數=" + mRetryCount);
                        if (mRetryCount > 0) {
                            return Observable.error(new Throwable(ERROR_RETRY));
                        } else if (ERROR_TOKEN.equals(throwable.getMessage())) {
                            mRetryCount++;
                            return TokenLoader.getInstance().getNetTokenLocked();
                        } else {
                            return Observable.error(throwable);
                        }
                    }
                });
            }
        });
        DisposableObserver<String> observer = new DisposableObserver<String>() {

            @Override
            public void onNext(String value) {
                Log.d(TAG, index + ":" + "收到信息=" + value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, index + ":" + "onError=" + e);
            }

            @Override
            public void onComplete() {
                Log.d(TAG, index + ":" + "onComplete");
            }
        };
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(observer);
    }
複製代碼

爲了方便你們閱讀，我把全部的邏輯都寫在了一整個調用鏈裏，整個調用鏈分爲四個部分：

• defer：讀取緩存中的token信息，這裏調用了TokenLoader中讀取緩存的接口，而這裏使用defer操做符，是爲了在重訂閱時，從新建立一個新的Observable，以讀取最新的緩存token信息，其原理圖以下：

  

• flatMap：經過token信息，請求必要的接口。
• retryWhen：使用重訂閱的方式來處理token失效時的邏輯，這裏分爲三種狀況：重試次數到達，那麼放棄重訂閱，直接返回錯誤；請求token接口，根據token請求的結果決定是否重訂閱；其它狀況直接放棄重訂閱。
• subscribe：返回接口數據。

2.4 TokenLoader 的實現

關鍵點在於TokenLoader的實現邏輯，代碼以下：

public class TokenLoader {

    private static final String TAG = TokenLoader.class.getSimpleName();

    private AtomicBoolean mRefreshing = new AtomicBoolean(false);
    private PublishSubject<String> mPublishSubject;
    private Observable<String> mTokenObservable;

    private TokenLoader() {
        mPublishSubject = PublishSubject.create();
        mTokenObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Thread.sleep(1000);
                Log.d(TAG, "發送Token");
                e.onNext(String.valueOf(System.currentTimeMillis()));
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String token) throws Exception {
                Log.d(TAG, "存儲Token=" + token);
                Store.getInstance().setToken(token);
                mRefreshing.set(false);
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                mRefreshing.set(false);
            }
        }).subscribeOn(Schedulers.io());
    }

    public static TokenLoader getInstance() {
        return Holder.INSTANCE;
    }

    private static class Holder {
        private static final TokenLoader INSTANCE = new TokenLoader();
    }

    public String getCacheToken() {
        return Store.getInstance().getToken();
    }

    public Observable<String> getNetTokenLocked() {
        if (mRefreshing.compareAndSet(false, true)) {
            Log.d(TAG, "沒有請求，發起一次新的Token請求");
            startTokenRequest();
        } else {
            Log.d(TAG, "已經有請求，直接返回等待");
        }
        return mPublishSubject;
    }

    private void startTokenRequest() {
        mTokenObservable.subscribe(mPublishSubject);
    }

}
複製代碼

在retryWhen中，咱們調用了getNetTokenLocked來得到一個PublishSubject，爲了實現前面說到的下面這個邏輯：

咱們使用了一個 AtomicBoolean來標記是否有刷新 Token的請求正在執行，若是有，那麼直接返回一個 PublishSubject，不然就先發起一次刷新 token的請求，並將 PublishSubject做爲該請求的訂閱者。

這裏用到了PublishSubject的特性，它既是做爲Token請求的訂閱者，同時又做爲retryWhen函數所返回Observable的發送方，由於retryWhen返回的Observable所發送的值就決定了是否須要重訂閱：

• 若是Token請求返回正確，那麼就會發送onNext事件，觸發重訂閱操做，使得咱們能夠再次觸發一次重試操做。
• 若是Token請求返回錯誤，那麼就會放棄重訂閱，使得整個請求的調用鏈結束。

而AtomicBoolean保證了多線程的狀況下，只能有一個刷新Token的請求，在這個階段內不會觸發重複的刷新token請求，僅僅是做爲觀察者而已，而且能夠在刷新token的請求回來以後馬上進行重訂閱的操做。在doOnNext/doOnError中，咱們將正在刷新的標誌位恢復，同時緩存最新的token。

爲了模擬上面提到的多線程請求刷新token的狀況，咱們在發起一個請求500ms以後，馬上發起另外一個請求，當第二個請求決定是否要重訂閱時，第一個請求正在進行刷新token的操做。

@Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_token);
        mBtnRequest = (Button) findViewById(R.id.bt_request);
        mBtnRequest.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                startRequest(0);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                startRequest(1);
            }
        });
    }
複製代碼

控制檯的輸出以下，能夠看到在第二個請求決定是否要重訂閱時，它判斷到已經有請求，所以只是等待而已。而在第一個請求致使的token刷新回調以後，兩個請求都進行了重試，併成功地請求到了接口信息。

2.5 操做符

本文中用到的操做符的官方解釋連接以下：

• Create
• Defer
• FlatMap
• RetryWhen
• PublishSubject

關於retryWhen的更詳細的解釋，推薦你們能夠看一下以前的 RxJava2 實戰知識梳理(6) - 基於錯誤類型的重試請求，它是這篇文章的基礎。

---

更多文章，歡迎訪問個人 Android 知識梳理系列：

• Android 知識梳理目錄：www.jianshu.com/p/fd82d1899…
• 我的主頁：lizejun.cn
• 我的知識總結目錄：lizejun.cn/categories/