Android使用WebSocket長連接實現消息推送
- 2020 年 4 月 8 日
- 筆記
前言
消息推送功能可以說移動APP不可缺少的功能之一，一般簡單的推送我們可以使用第三方推送的SDK，比如極光推送、信鴿推送等，但是對於消息聊天這種及時性有要求的或者三方推送不滿足業務需求的，我們就需要使用WebSocket實現消息推送功能。
基本流程
WebSocket是什麼，這裡就不做介紹了，我們這裡使用的開源框架是https://github.com/TakahikoKawasaki/nv-websocket-client
基於開源協議我們封裝實現WebSocket的連接、註冊、心跳、消息分發、超時任務功能，基本流程如下：
連接功能
首先我們新建一個項目，在build.grade中添加配置
compile 'com.neovisionaries:nv-websocket-client:2.2'
新建websocket管理類WsManger
public class WsManager { private volatile static WsManager wsManger; private WsManager() { } public static WsManager getWsManger() { if (wsManger == null) { synchronized (WsManager.class) { if (wsManger == null) { wsManger = new WsManager(); } } } return wsManger; } }
接下來添加連接方法，我們將webSocket的狀態分為三種，新建WsStatue枚舉類對應起來
public enum WsStatus { /** * 連接成功 */ CONNECT_SUCCESS, /** * 連接失敗 */ CONNECT_FAIL, /** * 正在連接 */ CONNECTING; }
連接方法如下所示：
/** * 連接方法 這裡要判斷是否登錄 此處省略 */ public void connect() { //WEB_SOCKET_API 是連接的url地址， // CONNECT_TIMEOUT是連接的超時時間 這裡是 5秒 try { ws = new WebSocketFactory().createSocket(WEB_SOCKET_API, CONNECT_TIMEOUT) //設置幀隊列最大值為5 .setFrameQueueSize(5) //設置不允許服務端關閉連接卻未發送關閉幀 .setMissingCloseFrameAllowed(false) //添加回調監聽 .addListener(new WsListener()) //非同步連接 .connectAsynchronously(); } catch (IOException e) { e.printStackTrace(); } setStatus(WsStatus.CONNECTING); }
調用連接方法後 我們來看連接的回調 也就是WsListener
/** * websocket回調事件 */ private class WsListener extends WebSocketAdapter { @Override public void onConnected(WebSocket websocket, Map<String, List<String>> headers) throws Exception { Log.d(TAG, "onConnected: 連接成功"); } @Override public void onConnectError(WebSocket websocket, WebSocketException exception) throws Exception { Log.d(TAG, "onConnectError: 連接失敗"); } @Override public void onDisconnected(WebSocket websocket, WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer) throws Exception { Log.d(TAG, "onDisconnected: 斷開連接"); } @Override public void onTextMessage(WebSocket websocket, String text) throws Exception { Log.d(TAG, "onTextMessage: 收到消息:" + text); } }
下面我們調用連接方法
WsManager.getWsManger().connect();
運行項目我們可以看到如下列印：
此處我們要做的處理是，如果收到連接失敗或者斷開連接的回調 需要重新連接，我們重新調用一次連接方法即可，並且如果超過三次重連失敗，我們在業務中可以通過調用介面來獲取數據，避免數據丟失，此處細節省略。
協議封裝
此處協議如下所示：
{ "action":"", "requestChild":{ "clientType":"", "id":"" } }
心跳、發送請求都屬於客戶端主動發送請求，對於請求結果我們分為成功和失敗以及超時，發送超時我們是收不到伺服器任何回復的，所以我們需要在發送之後將發送放在超時任務隊列中，如果請求成功將任務從超時隊列中移除，超時從超時隊列中獲取任務重新請求。
超時任務隊列中回調有成功、失敗、超時。
我們按照上述協議，新增對應實體類，採用Builder設計模式
public class Request { /** * 行為 */ private String action; /** * 請求體 */ private RequestChild req; /** * 請求次數 */ private transient int reqCount; /** * 超時的時間 */ private transient int timeOut; public Request() { } public Request(String action, int reqCount, int timeOut, RequestChild req) { this.action = action; this.req = req; this.reqCount = reqCount; this.timeOut = timeOut; } public static class Builder { //action 請求類型 private String action; //請求子類數據 按照具體業務劃分 private RequestChild req; //請求次數 便於重試 private int reqCount; //超時時間 private int timeOut; public Builder action(String action) { this.action = action; return this; } public Builder req(RequestChild req) { this.req = req; return this; } public Builder reqCount(int reqCount) { this.reqCount = reqCount; return this; } public Builder timeOut(int timeOut) { this.timeOut = timeOut; return this; } public Request build() { return new Request(action, reqCount, timeOut, req); } } }
public class RequestChild { /** * 設備類型 */ private String clientType; /** * 用於用戶註冊的id */ private String id; public RequestChild(String clientType, String id) { this.clientType = clientType; this.id = id; } public RequestChild() { } public static class Builder { private String clientType; private String id; public RequestChild.Builder setClientType(String clientType) { this.clientType = clientType; return this; } public RequestChild.Builder setId(String id) { this.id = id; return this; } public RequestChild build() { return new RequestChild(clientType, id); } } }
我們添加一個發送請求的方法如下：
/** * 發送請求 * * @param request 請求體 * @param reqCount 請求次數 * @param requestListern 請求回調 */ private void senRequest(Request request, final int reqCount, final RequestListern requestListern) { if (!isNetConnect()) { requestListern.requestFailed("網路未連接"); return; } }
請求回調如下所示
public interface RequestListern { /** * 請求成功 */ void requestSuccess(); /** * 請求失敗 * * @param message 請求失敗消息提示 */ void requestFailed(String message); }
接著我們要把請求放在超時隊列中，新建超時任務類，對應的分別是請求參數、請求回調、任務調度
public class TimeOutTask { /** * 請求主體 */ private Request request; /** * 通用返回 */ private RequestCallBack requestCallBack; /** * r任務 */ private ScheduledFuture scheduledFuture; public TimeOutTask(Request request, RequestCallBack requestCallBack, ScheduledFuture scheduledFuture) { this.request = request; this.requestCallBack = requestCallBack; this.scheduledFuture = scheduledFuture; } public ScheduledFuture getScheduledFuture() { return scheduledFuture; } public void setScheduledFuture(ScheduledFuture scheduledFuture) { this.scheduledFuture = scheduledFuture; } public Request getRequest() { return request; } public void setRequest(Request request) { this.request = request; } public RequestCallBack getRequestCallBack() { return requestCallBack; } public void setRequestCallBack(RequestCallBack requestCallBack) { this.requestCallBack = requestCallBack; } }
RequestCallBack是超時任務的回調，只是比請求回調多了個超時，因為超時的處理機制是一樣的，所以這裡我們沒必要將超時回調到請求中
public interface RequestCallBack { /** * 請求成功 */ void requestSuccess(); /** * 請求失敗 * * @param request 請求體 * @param message 請求失敗的消息 */ void requestFailed(String message, Request request); /** * 請求超時 * * @param request 請求體 */ void timeOut(Request request); }
/** * 添加超時任務 */ private ScheduledFuture enqueueTimeout(final Request request, final long timeout) { Log.d(TAG, " " + "enqueueTimeout: 添加超時任務類型為：" + request.getAction()); return executor.schedule(new Runnable() { @Override public void run() { TimeOutTask timeoutTask = callbacks.remove(request.getAction()); if (timeoutTask != null) { timeoutTask.getRequestCallBack().timeOut(timeoutTask.getRequest()); } } }, timeout, TimeUnit.MILLISECONDS); }
超時任務的方法 是通過任務調度定時調用，請求成功後我們會把超時任務移除,當到了超時時間時，任務還存在就說明任務超時了。
每次的任務我們以action為鍵值存在hashMap中
private Map<String, CallbackWrapper> callbacks = new HashMap<>();
將任務放入超時任務程式碼如下所示：
final ScheduledFuture timeoutTask = enqueueTimeout(request, request.getTimeOut()); final RequestCallBack requestCallBack = new RequestCallBack() { @Override public void requestSuccess() { requestListern.requestSuccess(); } @Override public void requestFailed(String message, Request request) { requestListern.requestFailed(message); } @Override public void timeOut(Request request) { timeOutHanlder(request); } }; callbacks.put(request.getAction(), new CallbackWrapper(request, requestCallBack, timeoutTask));
一般而言，任務超時都是由於連接原因導致，所以我們這裡可以嘗試重試一次，如果還是超時，通過 timeOutHanlder(request);方法 進行重新連接，重連程式碼和連接程式碼一樣，這裡就省略了，做好這步操作，我們就可以發送消息了。
/** * 超時任務 */ private void timeOutHanlder(Request requset) { setStatus(WsStatus.CONNECT_FAIL); //這裡假裝有重連 Log.d(TAG, "timeOutHanlder: 請求超時 準備重連"); }
到這裡我們的流程基本可以走通了。
心跳
首先我們要了解下心跳的作用是什麼，心跳是在連接成功後，通過固定的間隔時間向伺服器發送詢問，當前是否還在線，有很多人說心跳失敗我們就重連，成功就繼續心跳，但是這裡要注意的是，我們一般是收不到心跳失敗回調的，心跳也是向伺服器發送數據，所以我們要將所有的主動請求都放在超時任務隊列中，
所以對websocket來說 請求結果有三種：成功、失敗、超時，對於用戶 只有成功、失敗即可。
至於心跳、註冊等請求發送的數據是什麼，這就得看我們與服務端定的協議是什麼樣了，通常來說 分為action 和 requestBody，協議格式我們再第二步已經封裝好了，這裡我們以心跳任務為例驗證上面的封裝。
/** * 心跳 */ void keepAlive() { Request request = new Request.Builder() .reqCount(0) .timeOut(REQUEST_TIMEOUT) .action(ACTION_KEEPALIVE).build(); WsManager.getWsManger().senRequest(request, request.getReqCount() + 1, new RequestListern() { @Override public void requestSuccess() { Log.d(TAG, "requestSuccess: 心跳發送成功了"); } @Override public void requestFailed(String message) { } }); }
我們每間隔10s中開啟一次心跳任務
/** * 開始心跳 */ public void startKeepAlive() { mHandler.postDelayed(mKeepAliveTask, HEART_BEAT_RATE); }
/** * 心跳任務 */ private Runnable mKeepAliveTask = new Runnable() { @Override public void run() { keepAlive(); mHandler.removeCallbacks(mKeepAliveTask); mHandler.postDelayed(mKeepAliveTask, HEART_BEAT_RATE); } };
為了便於操作演示，在主頁面上加個按鈕 ，點擊按鈕調用startKeepAlive方法，運行如下所示：
我們可以看到心跳返回的statue是300 不成功，5秒之後走到了請求超時的方法中，所以如果狀態返回成功的話，我們需要回調給調用者
/** * 處理 任務回調 * * @param action 請求類型 */ void disPatchCallbackWarp(String action, boolean isSuccess) { CallbackWrapper callBackWarp = callbacks.remove(action); if (callBackWarp == null) { Logger.d(TAG+" "+ "disPatchCallbackWarp: 任務隊列為空"); } else { callBackWarp.getScheduledFuture().cancel(true); if (isSuccess) { callBackWarp.getRequestCallBack().requestSuccess(); } else { callBackWarp.getRequestCallBack().requestFailed("", new Request()); } } }
這樣調用者才知道成功或失敗。
發送其他消息與心跳一樣，只是請求參數不同而已，修改Request參數即可。這樣我們根據協議和業務就實現一個比較規範的webSocket消息推送流程了。
