在android上如何使用WebSocket做消息通信功能

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;
 }
 
 
}

public enum WsStatus {
 
 /**
 * 連接成功
 */
 CONNECT_SUCCESS,
 /**
 * 連接失敗
 */
 CONNECT_FAIL,
 /**
 * 正在連接
 */
 CONNECTING;
}

/**
 * 連接方法 這里要判斷是否登錄 此處省略
 */
public void connect() {
 //WEB_SOCKET_API 是連接的url地址，
 // CONNECT_TIMEOUT是連接的超時(shí)時(shí)間 這里是 5秒
 try {
 ws = new WebSocketFactory().createSocket(WEB_SOCKET_API, CONNECT_TIMEOUT)
 //設(shè)置幀隊(duì)列最大值為5
 .setFrameQueueSize(5)
 //設(shè)置不允許服務(wù)端關(guān)閉連接卻未發(fā)送關(guān)閉幀
 .setMissingCloseFrameAllowed(false)
 //添加回調(diào)監(jiān)聽(tīng)
 .addListener(new WsListener())
 //異步連接
 .connectAsynchronously();
 } catch (IOException e) {
 e.printStackTrace();
 }
 setStatus(WsStatus.CONNECTING);
}

/**
 * websocket回調(diào)事件
 */
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: 斷開(kāi)連接");
 
 }
 
 @Override
 public void onTextMessage(WebSocket websocket, String text) throws Exception {
 Log.d(TAG, "onTextMessage: 收到消息:" + text);
 }
}

WsManager.getWsManger().connect();

{
 "action":"",
 "requestChild":{
 "clientType":"",
 "id":""
 }
}

public class Request {
 
 /**
 * 行為
 */
 private String action;
 
 /**
 * 請(qǐng)求體
 */
 private RequestChild req;
 
 
 /**
 * 請(qǐng)求次數(shù)
 */
 private transient int reqCount;
 
 /**
 * 超時(shí)的時(shí)間
 */
 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 請(qǐng)求類(lèi)型
 private String action;
 //請(qǐng)求子類(lèi)數(shù)據(jù) 按照具體業(yè)務(wù)劃分
 private RequestChild req;
 //請(qǐng)求次數(shù) 便于重試
 private int reqCount;
 //超時(shí)時(shí)間
 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 {
 
 /**
 * 設(shè)備類(lèi)型
 */
 private String clientType;
 
 
 /**
 * 用于用戶注冊(cè)的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);
 }
 
 }
 
 
}

/**
 * 發(fā)送請(qǐng)求
 *
 * @param request 請(qǐng)求體
 * @param reqCount 請(qǐng)求次數(shù)
 * @param requestListern 請(qǐng)求回調(diào)
 */
private void senRequest(Request request, final int reqCount, final RequestListern requestListern) {
 if (!isNetConnect()) {
 requestListern.requestFailed("網(wǎng)絡(luò)未連接");
 return;
 }
 
}

public interface RequestListern {
 
 /**
 * 請(qǐng)求成功
 */
 void requestSuccess();
 
 /**
 * 請(qǐng)求失敗
 *
 * @param message 請(qǐng)求失敗消息提示
 */
 void requestFailed(String message);
}

public class TimeOutTask {
 
 
 /**
 * 請(qǐng)求主體
 */
 private Request request;
 
 /**
 * 通用返回
 */
 private RequestCallBack requestCallBack;
 
 /**
 * r任務(wù)
 */
 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;
 }
 
}

public interface RequestCallBack {
 
 /**
 * 請(qǐng)求成功
 */
 void requestSuccess();
 
 /**
 * 請(qǐng)求失敗
 *
 * @param request 請(qǐng)求體
 * @param message 請(qǐng)求失敗的消息
 */
 void requestFailed(String message, Request request);
 
 /**
 * 請(qǐng)求超時(shí)
 *
 * @param request 請(qǐng)求體
 */
 void timeOut(Request request);
}
/**
 * 添加超時(shí)任務(wù)
 */
private ScheduledFuture enqueueTimeout(final Request request, final long timeout) {
 Log.d(TAG, " " + "enqueueTimeout: 添加超時(shí)任務(wù)類(lèi)型為：" + 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);
}

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));

/**
 * 超時(shí)任務(wù)
 */
private void timeOutHanlder(Request requset) {
 setStatus(WsStatus.CONNECT_FAIL);
 //這里假裝有重連
 Log.d(TAG, "timeOutHanlder: 請(qǐng)求超時(shí) 準(zhǔn)備重連");
}

/**
 * 心跳
 */
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: 心跳發(fā)送成功了");
 }
 
 @Override
 public void requestFailed(String message) {
 }
 });
}

/**
 * 開(kāi)始心跳
 */
public void startKeepAlive() {
 mHandler.postDelayed(mKeepAliveTask, HEART_BEAT_RATE);
}
/**
 * 心跳任務(wù)
 */
private Runnable mKeepAliveTask = new Runnable() {
 
 @Override
 public void run() {
 keepAlive();
 mHandler.removeCallbacks(mKeepAliveTask);
 mHandler.postDelayed(mKeepAliveTask, HEART_BEAT_RATE);
 }
};

/**
 * 處理 任務(wù)回調(diào)
 *
 * @param action 請(qǐng)求類(lèi)型
 */
void disPatchCallbackWarp(String action, boolean isSuccess) {
 CallbackWrapper callBackWarp = callbacks.remove(action);
 if (callBackWarp == null) {
 Logger.d(TAG+" "+ "disPatchCallbackWarp: 任務(wù)隊(duì)列為空");
 } else {
 callBackWarp.getScheduledFuture().cancel(true);
 if (isSuccess) {
 callBackWarp.getRequestCallBack().requestSuccess();
 } else {
 callBackWarp.getRequestCallBack().requestFailed("", new Request());
 }
 
 }
}