[[414380]]

想了解更多內容，請訪問：

51CTO和華為官方合作共建的鴻蒙技術社區

https://harmonyos.51cto.com

一. 概述

RxWebSocket是一個基于okhttp和RxJava封裝的WebSocket客戶端,此庫的核心特點是 除了手動關閉WebSocket(就是RxJava取消訂閱),WebSocket在異常關閉的時候(onFailure,發生異常,如WebSocketException等等),會自動重連,永不斷連.其次,對WebSocket做的緩存處理,同一個URL,共享一個WebSocket.

由于是基于RxJava封裝,所以帶來了無限可能,可以和RxBinding,Rxlifecycle一起使用,方便對WebSocket的管理.

效果圖

項目已經上傳Jcenter,依賴方法:

//本項目 
compile 'io.github.dzsf:RxWebSocket:1.0.0' 

二. 使用方法

0.初始化,可以也忽略直接使用.

如果你想使用自己的okhttpClient:

OkHttpClient yourClient = new OkHttpClient(); 
RxWebSocketUtil.getInstance().setClient(yourClient); 

 是否打印日志:

RxWebSocketUtil.getInstance().setShowLog(BuildConfig.DEBUG); 

1.獲取一個WebSocket,接收消息,多種方式:

RxWebSocketUtil.getInstance().getWebSocketInfo(url) 
        .subscribe(new Action1<WebSocketInfo>() { 
            @Override 
            public void call(WebSocketInfo webSocketInfo) { 
                mWebSocket = webSocketInfo.getWebSocket(); 
                Log.d("MainActivity", webSocketInfo.getString()); 
                Log.d("MainActivity", "ByteString:" + webSocketInfo.getByteString()); 
            } 
        }); 
 
mWebSocket.send("hello word"); 
 
//get StringMsg 
RxWebSocketUtil.getInstance().getWebSocketString(url) 
        .subscribe(new Action1<String>() { 
            @Override 
            public void call(String s) { 
            } 
        }); 
// get ByteString 
RxWebSocketUtil.getInstance().getWebSocketByteString(url) 
        .subscribe(new Action1<ByteString>() { 
            @Override 
            public void call(ByteString byteString) { 
            } 
        }); 
//get WebSocket 
RxWebSocketUtil.getInstance().getWebSocket(url) 
        .subscribe(new Action1<WebSocket>() { 
            @Override 
            public void call(WebSocket webSocket) { 
            } 
        }); 
// 帶timeout的WebSocket,當在指定時間內沒有收到消息,就重連WebSocket.為了適配小米平板. 
RxWebSocketUtil.getInstance().getWebSocketInfo(url,10, TimeUnit.SECONDS) 
        .subscribe(new Action1<WebSocketInfo>() { 
            @Override 
            public void call(WebSocketInfo webSocketInfo) { 
            } 
        }); 

2.發送消息:

//用WebSocket的引用直接發 
mWebSocket.send("hello word"); 
 
//url 對應的WebSocket 必須打開,否則報錯 
RxWebSocket.send(sendUrl, "hello"); 
RxWebSocket.send(sendUrl, ByteString.EMPTY); 
 
//異步發送,若WebSocket已經打開,直接發送,若沒有打開,打開一個WebSocket發送完數據,直接關閉. 
RxWebSocket.asyncSend(sendUrl, "hello"); 
RxWebSocket.asyncSend(sendUrl, ByteString.EMPTY); 

3.關閉WebSocket:

項目是依托RxJava實現的,所以關閉WebSocket的方法也就是在適當的時候注銷 Observable,項目里的demo里,寫了一個簡單的lifecycle,將Observable生命綁定到Activity的onDestroy,自動注銷.代碼細節請看demo,因為內部實現了同一個URL的WebSocket共享機制,所以當外部所有持有這個URL的Observable都注銷后,這個WebSocket連接就會自動關閉.請看原理解析部分

//注意取消訂閱,有多種方式,比如 rxlifecycle 
mSubscription = RxWebSocketUtil.getInstance().getWebSocketInfo(url) 
        .subscribe(new Action1<WebSocketInfo>() { 
            @Override 
            public void call(WebSocketInfo webSocketInfo) { 
                mWebSocket = webSocketInfo.getWebSocket(); 
 
                if (webSocketInfo.isOnOpen()) { 
                    Log.d("MainActivity", " on WebSocket open"); 
                } else { 
                    String string = webSocketInfo.getString(); 
 
                    if (string != null) { 
                        Log.d("MainActivity", string); 
                        textview.setText(Html.fromHtml(string)); 
                    } 
 
                    ByteString byteString = webSocketInfo.getByteString(); 
 
                    if (byteString != null) { 
                        Log.d("MainActivity", 
                            "webSocketInfo.getByteString():" + 
                            byteString); 
                    } 
                } 
            } 
        }); 
 
//注銷 
if (mSubscription != null) { 
    mSubscription.unsubscribe(); 
} 
 
//lifecycle注銷,詳情看demo 
RxWebSocketUtil.getInstance().getWebSocketString(url) 
            .compose(this.<String>bindOnActivityEvent(ActivityEvent.onDestory)) 
            .subscribe(new Action1<String>() { 
                @Override 
                public void call(String s) { 
                } 
            }); 

三. 原理解析

1. 首先需要將okhttp的WebSocket包裝成Observable,由于需要將WebSocket,Stringmsg,ByteString等信息一同發送給觀察者所以先構建一個WebSocketInfo類,將信息封裝:

public class WebSocketInfo { 
    private WebSocket mWebSocket; 
    private String mString; 
    private ByteString mByteString; 
    private boolean onOpen; 
    //其他省略 
} 

onOpen字段主要用來判斷當前的這個WebSocketInfo是否是當WebSocket打開時發送的消息(onOpen),這時,Stringmsg和ByteString都是null.

2. 將WebSocketInfo包裝成Observable發出:

private final class WebSocketOnSubscribe implements Observable.OnSubscribe<WebSocketInfo> { 
    private String url; 
    private WebSocket webSocket; 
    private WebSocketInfo startInfo; 
    private WebSocketInfo stringInfo; 
    private WebSocketInfo byteStringInfo; 
 
    public WebSocketOnSubscribe(String url) { 
        this.url = url; 
        startInfo = new WebSocketInfo(true); 
        stringInfo = new WebSocketInfo(); 
        byteStringInfo = new WebSocketInfo(); 
    } 
 
    @Override 
    public void call(final Subscriber<?super WebSocketInfo> subscriber) { 
        if (webSocket != null) { 
            //降低重連頻率 
            if (!"main".equals(Thread.currentThread().getName())) { 
                SystemClock.sleep(2000); 
            } 
        } 
 
        initWebSocket(subscriber); 
    } 
 
    private void initWebSocket( 
        final Subscriber<?super WebSocketInfo> subscriber) { 
        webSocket = client.newWebSocket(getRequest(url), 
                new WebSocketListener() { 
                    @Override 
                    public void onOpen(final WebSocket webSocket, 
                        Response response) { 
                        if (showLog) { 
                            Log.d("RxWebSocketUtil", url + " --> onOpen"); 
                        } 
 
                        webSocketMap.put(url, webSocket); 
                        AndroidSchedulers.mainThread().createWorker().schedule(new Action0() { 
                                @Override 
                                public void call() { 
                                    if (!subscriber.isUnsubscribed()) { 
                                        subscriber.onStart(); 
                                        startInfo.setWebSocket(webSocket); 
                                        subscriber.onNext(startInfo); 
                                    } 
                                } 
                            }); 
                    } 
 
                    @Override 
                    public void onMessage(WebSocket webSocket, String text) { 
                        if (!subscriber.isUnsubscribed()) { 
                            stringInfo.setWebSocket(webSocket); 
                            stringInfo.setString(text); 
                            subscriber.onNext(stringInfo); 
                        } 
                    } 
 
                    @Override 
                    public void onMessage(WebSocket webSocket, ByteString bytes) { 
                        if (!subscriber.isUnsubscribed()) { 
                            byteStringInfo.setWebSocket(webSocket); 
                            byteStringInfo.setByteString(bytes); 
                            subscriber.onNext(byteStringInfo); 
                        } 
                    } 
 
                    @Override 
                    public void onFailure(WebSocket webSocket, Throwable t, 
                        Response response) { 
                        if (showLog) { 
                            Log.e("RxWebSocketUtil", 
                                t.toString() + 
                                webSocket.request().url().uri().getPath()); 
                        } 
 
                        if (!subscriber.isUnsubscribed()) { 
                            subscriber.onError(t); 
                        } 
                    } 
 
                    @Override 
                    public void onClosing(WebSocket webSocket, int code, 
                        String reason) { 
                        webSocket.close(1000, null); 
                    } 
 
                    @Override 
                    public void onClosed(WebSocket webSocket, int code, 
                        String reason) { 
                        if (showLog) { 
                            Log.d("RxWebSocketUtil", 
                                url + " --> onClosed:code= " + code); 
                        } 
                    } 
                }); 
        subscriber.add(new MainThreadSubscription() { 
                @Override 
                protected void onUnsubscribe() { 
                    webSocket.close(3000, "手動關閉"); 
                } 
            }); 
    } 
} 

實現一個WebSocketOnSubscribe 將WebSocket的回調轉化成subscriber調用.發送給Observable下游.在onOpen時調用 subscriber.onStart(),并且發送一個onOpen的WebSocketInfo.在subscriber注銷的時候關閉WebSocket.在call方法最上面有個SystemClock.sleep(2000),這個主要是為了降低在斷連的時候的重連頻率,將在下面講到.

包裝成Observable:

Observable.create(new WebSocketOnSubscribe(url)) 
            .subscribeOn(Schedulers.io()) 
            .observeOn(AndroidSchedulers.mainThread()); 

3. 實現自動重連:

Observable.create(new WebSocketOnSubscribe(url)) 
            //自動重連 
            .timeout(timeout, timeUnit).retry() 
            .subscribeOn(Schedulers.io()) 
            .observeOn(AndroidSchedulers.mainThread()); 

RxJava retry操作符,很完美的實現了這個功能,當上游發出Throwable的時候,retry將錯誤吃掉,并重新調用 onSubscribe的call方法,也就是WebSocketOnSubscribe的call,就會重新初始化一個WebSocket連接,達到重連的目的,如果一直沒有網絡,這個retry的調用頻率非常高,所以在call方法里面,當是重連的時候,就SystemClock.sleep(2000),休眠2秒,這樣重連的頻率就是2秒重連一次. 當然在retry上面還有一個timeout操作符.當subscriber.onNext()在指定時間間隔里沒有調用,就發出一個timeoutException,讓retry重連WebSocket.這個主要是為了適配部分國產機型,當WebSocket發生連接異常時,不會及時發出錯誤,如小米平板.在每次重連都會把原來的WebSocket關閉.

4. 實現同一個URL的WebSocket共享

Observable.create(new WebSocketOnSubscribe(url)) 
        //自動重連 
        .timeout(timeout, timeUnit) 
        .retry() 
        //共享 
        .doOnUnsubscribe(new Action0() { 
            @Override 
            public void call() { 
                observableMap.remove(url); 
                webSocketMap.remove(url); 
 
                if (showLog) { 
                    Log.d("RxWebSocketUtil", "注銷"); 
                } 
            } 
        }) 
        .doOnNext(new Action1<WebSocketInfo>() { 
            @Override 
            public void call(WebSocketInfo webSocketInfo) { 
                if (webSocketInfo.isOnOpen()) { 
                    webSocketMap.put(url, webSocketInfo.getWebSocket()); 
                } 
            } 
        }) 
        .share() 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread()); 

實現共享功能,主要是為了防止一個URL的WebSocket,建立多個連接,這個主要是由RxJava的share操作符實現,share操作符,使得一個Observable可以有多個subscriber,當有多個subscriber時,當所有的subscriber都取消訂閱,這個Observable才會取消訂閱.

getWebSocketInfo()方法完整代碼:

public Observable<WebSocketInfo> getWebSocketInfo(final String url, final long timeout, final TimeUnit timeUnit) { 
    Observable<WebSocketInfo> observable = observableMap.get(url); 
    if (observable == null) { 
        observable = Observable.create(new WebSocketOnSubscribe(url)) 
                //自動重連 
                .timeout(timeout, timeUnit) 
                .retry() 
                //共享 
                .doOnUnsubscribe(new Action0() { 
                    @Override 
                    public void call() { 
                        observableMap.remove(url); 
                        webSocketMap.remove(url); 
                        if (showLog) { 
                            Log.d("RxWebSocketUtil", "注銷"); 
                        } 
                    } 
                }) 
                .doOnNext(new Action1<WebSocketInfo>() { 
                    @Override 
                    public void call(WebSocketInfo webSocketInfo) { 
                        if (webSocketInfo.isOnOpen()) { 
                            webSocketMap.put(url, webSocketInfo.getWebSocket()); 
                        } 
                    } 
                }) 
                .share() 
                .subscribeOn(Schedulers.io()) 
                .observeOn(AndroidSchedulers.mainThread()); 
        observableMap.put(url, observable); 
    } else { 
        observable = Observable.merge(Observable.just(new WebSocketInfo(webSocketMap.get(url), true)), observable); 
    } 
    return observable; 
} 

doOnUnsubscribe作用:在Observable注銷,即 WebSocket關閉時,移除map中的緩存的Observable和WebSocket.

doOnNext作用: 判斷接收到的WebSocketInfo是否是WebSocket在onOpen的時候發的,然后將其緩存起來.作用就是:如果有一個相同的URL訂閱Observable,就從緩存中取,這個時候我們應該把一個WebSocket的onOpen事件也發給這個訂閱者:

//使用merge操作符,將onOpen事件發給訂閱者 
observable = Observable.merge(Observable.just(new WebSocketInfo(webSocketMap.get(url), true)), observable); 

這樣的話,同一個URL的WebSocket,不管在什么地方什么時間訂閱,都能收到一個onOpen事件,外部表現的就像一個新的WebSocket.

getWebSocketInfo方法的幾種變體:

/** 
 * default timeout: 30 days 
 * <p> 
 * 若忽略小米平板,請調用這個方法 
 * </p> 
 */ 
public Observable<WebSocketInfo> getWebSocketInfo(String url) { 
    return getWebSocketInfo(url, 30, TimeUnit.DAYS); 
} 
 
public Observable<String> getWebSocketString(String url) { 
    return getWebSocketInfo(url) 
            .map(new Func1<WebSocketInfo, String>() { 
                @Override 
                public String call(WebSocketInfo webSocketInfo) { 
                    return webSocketInfo.getString(); 
                } 
            }) 
            .filter(new Func1<String, Boolean>() { 
                @Override 
                public Boolean call(String s) { 
                    return s != null; 
                } 
            }); 
} 
 
public Observable<ByteString> getWebSocketByteString(String url) { 
    return getWebSocketInfo(url) 
            .map(new Func1<WebSocketInfo, ByteString>() { 
                @Override 
                public ByteString call(WebSocketInfo webSocketInfo) { 
                    return webSocketInfo.getByteString(); 
                } 
            }) 
            .filter(new Func1<ByteString, Boolean>() { 
                @Override 
                public Boolean call(ByteString byteString) { 
                    return byteString != null; 
                } 
            }); 
} 
 
public Observable<WebSocket> getWebSocket(String url) { 
    return getWebSocketInfo(url) 
            .map(new Func1<WebSocketInfo, WebSocket>() { 
                @Override 
                public WebSocket call(WebSocketInfo webSocketInfo) { 
                    return webSocketInfo.getWebSocket(); 
                } 
            }); 
} 

5 . send信息到服務端

上面已經講到WebSocketInfo包含了WebSocket,所以在訂閱后,就可以拿到這個WebSocket引用就可以WebSocket.send發送消息到服務端.當然我們的RxWebSocketUtil已經將開啟的WebSocket已經緩存.所以我們也可以這樣發消息:

/** 
 * 如果url的WebSocket已經打開,可以直接調用這個發送消息. 
 * 
 * @param url 
 * @param msg 
 */ 
public void send(String url, String msg) { 
    WebSocket webSocket = webSocketMap.get(url); 
    if (webSocket != null) { 
        webSocket.send(msg); 
    } else { 
        throw new IllegalStateException("The WebSokcet not open"); 
    } 
} 
 
/** 
 * 如果url的WebSocket已經打開,可以直接調用這個發送消息. 
 * 
 * @param url 
 * @param byteString 
 */ 
public void send(String url, ByteString byteString) { 
    WebSocket webSocket = webSocketMap.get(url); 
    if (webSocket != null) { 
        webSocket.send(byteString); 
    } else { 
        throw new IllegalStateException("The WebSokcet not open"); 
    } 
} 

當指定的URL的WebSocket沒有打開會直接報錯.

異步發送消息到服務端

/** 
 * 不用關心url 的WebSocket是否打開,可以直接發送 
 * 
 * @param url 
 * @param msg 
 */ 
public void asyncSend(String url, final String msg) { 
    getWebSocket(url) 
            .first() 
            .subscribe(new Action1<WebSocket>() { 
                @Override 
                public void call(WebSocket webSocket) { 
                    webSocket.send(msg); 
                } 
            }); 
 
} 
 
/** 
 * 不用關心url 的WebSocket是否打開,可以直接發送 
 * 
 * @param url 
 * @param byteString 
 */ 
public void asyncSend(String url, final ByteString byteString) { 
    getWebSocket(url) 
            .first() 
            .subscribe(new Action1<WebSocket>() { 
                @Override 
                public void call(WebSocket webSocket) { 
                    webSocket.send(byteString); 
                } 
            }); 
} 

這兩種發送方式,你不用關心URL的WebSocket是否打開,可以直接發送.實現思路也很簡單,getWebSocket(url)會獲取到Observable,或者是從緩存中取,或者是重新開啟一個WebSocket,但你都不需要關心,經過first操作符后,如果是從緩存取的Observable,就注銷的當前的Observable,當是新開的WebSocket,注銷掉當前的subscriber后,就沒有其他subscriber了,這個新開的WebSocket就會關閉(share操作符作用).

想了解更多內容，請訪問：

51CTO和華為官方合作共建的鴻蒙技術社區

https://harmonyos.51cto.com