前言
消息推送功能可以说移动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> 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中
京公网安备 11010802041100号