devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Android WebSocket原理及实战（二）_android client.newwebsocket 发送消息

作者：我家小花儿 | 2024-08-28 07:57:33

赞

踩

android client.newwebsocket 发送消息

上期原理篇我们讲到了WebSocket和HTTP的区别及其优缺点，这期主要是实战部分，大多数Android项目不需要使用第三方重量级的WebSocket框架，所以我以OKHttp3.0自带的WebSocket为例，来进行讲解

添加依赖

implementation "com.squareup.okhttp3:okhttp:4.9.0"

实现步骤

  • 构建OkHttpClient配置初始化一些参数。

  • 使用WebSocket的Url地址连接。

  • 设置WebSocket的连接状态回调和消息回调。

  • 解析消息json处理业务等。

  • 连接成功后，使用WebSocket发送消息

1、配置OKHttpClient

  1. val mHolder = OkHttpClient.Builder()
  2.             .connectTimeout(CONNECT_TIME_OUT, TimeUnit.SECONDS)//设置读取超时时间
  3.             .readTimeout(READ_TIME_OUT, TimeUnit.SECONDS)//设置连接超时时间
  4.             .writeTimeout(3, TimeUnit.SECONDS)//设置写的超时时间
  5.             .build()

2、使用URL 构建WebSocket请求

  1.         // 构建鉴权url
  2.         val authUrl = TTSTools.createAuthUrl(mTTSConfig)
  3.         // 获取请求
  4.         val request: Request = Request.Builder().url(authUrl).build()

鉴权url 一般是ws[s]://host/servername/path?形式，至于ws和wss的区别，可以类比http和https,这里不做赘述

3、发起连接，配置回调

  • onOpen()，连接成功
  • onMessage(String text)，收到字符串类型的消息，一般我们都是使用这个
  • onMessage(ByteString bytes)，收到字节数组类型消息，我这里没有用到
  • onClosed()，连接关闭
  • onFailure()，连接失败，一般都是在这里发起重连操作
  1.         // 开始连接
  2.         val client: OkHttpClient = CusOkHttpClient.mInstance
  3.         mWebSocket = client.newWebSocket(
  4.             request,
  5.             object : WebSocketListener() {
  6.  
  7.                 override fun onOpen(webSocket: WebSocket, response: Response) {
  8.                     Logger.d(mTAG, "WebSocket-握手协议")
  9.                 }
  10.  
  11.                 override fun onMessage(webSocket: WebSocket, text: String) {
  12.                     Logger.d(mTAG, "WebSocket-返回结果}")
  13.                 }
  14.  
  15.                 override fun onClosed(webSocket: WebSocket, code: Int, reason: String)
  16.                 {
  17.                     Logger.d(mTAG, "WebSocket关闭")
  18.                 }
  19.  
  20.                 override fun onFailure(webSocket: WebSocket, 
  21.                 t: Throwable, response: Response?) {
  22.                     Logger.d(mTAG, "WebSocket出错--${t.message}")
  23.                 }
  24.             }
  25.         )

4、使用WebSocket对象发送消息，msg为消息内容（一般是json，当然你也可以使用其他的，例如xml等），send方法会马上返回发送结果

  1. /**
  2.      * 发送数据
  3.      */
  4.     private fun sendSimpleData(
  5.         simpleData: String
  6.     ) {
  7.         if (!TextUtils.isEmpty(simpleData)) {
  8.             mTtsText = ttsText
  9.         }
  10.         mWebSocket?.send(getRequest(ttsText))
  11.     }

getRequest方法主要是生成相应的请求JSON,数据协议由两端决定

  1.     private fun getRequest(rawText: String): String {
  2.         return JSONObject().apply {
  3.             put("text",rawText)
  4.         }.toString()
  5.     }

可以看出，OKHTTP3的WebSocket使用非常的简单，而WebSocket可以基于RxJava或协程做二次封装，下面基于RxJava的封装，我找了个比较好的案例，供大家学习

基于RxJava封装

1、定义Api调用接口，外部只需要接触Api无无需关心内部实现逻辑

  1. public interface WebSocketWorker {
  2.     /**
  3.      * 获取连接，并返回观察对象
  4.      */
  5.     Observable<WebSocketInfo> get(String url);
  6.  
  7.     /**
  8.      * 设置一个超时时间，在指定时间内如果没有收到消息，会尝试重连
  9.      *
  10.      * @param timeout  超时时间
  11.      * @param timeUnit 超时时间单位
  12.      */
  13.     Observable<WebSocketInfo> get(String url, long timeout, TimeUnit timeUnit);
  14.  
  15.     /**
  16.      * 发送，url的WebSocket已打开的情况下使用，否则会抛出异常
  17.      *
  18.      * @param msg 消息，看具体和后端协商的格式，一般为json
  19.      */
  20.     Observable<Boolean> send(String url, String msg);
  21.  
  22.     /**
  23.      * 发送，同上
  24.      *
  25.      * @param byteString 信息类型为ByteString
  26.      */
  27.     Observable<Boolean> send(String url, ByteString byteString);
  28.  
  29.     /**
  30.      * 不关心WebSocket是否连接，直接发送
  31.      */
  32.     Observable<Boolean> asyncSend(String url, String msg);
  33.  
  34.     /**
  35.      * 同上，只是消息类型为ByteString
  36.      */
  37.     Observable<Boolean> asyncSend(String url, ByteString byteString);
  38.  
  39.     /**
  40.      * 关闭指定Url的连接
  41.      */
  42.     Observable<Boolean> close(String url);
  43.  
  44.     /**
  45.      * 马上关闭指定Url的连接
  46.      */
  47.     boolean closeNow(String url);
  48.  
  49.     /**
  50.      * 关闭当前所有连接
  51.      */
  52.     Observable<List<Boolean>> closeAll();
  54.     /**
  55.      * 马上关闭所有连接
  56.      */
  57.     void closeAllNow();
  58. }

2、构建者模式，大量的配置参数，我们先使用一个Builder类保存，再使用build()方法生成RxWebSocket对象

  1. public class RxWebSocketBuilder {
  2.     Context mContext;
  3.     /**
  4.      * 是否打印Log
  5.      */
  6.     boolean mIsPrintLog;
  7.     /**
  8.      * Log代理对象
  9.      */
  10.     Logger.LogDelegate mLogDelegate;
  11.     /**
  12.      * 支持外部传入OkHttpClient
  13.      */
  14.     OkHttpClient mClient;
  15.     /**
  16.      * 支持SSL
  17.      */
  18.     SSLSocketFactory mSslSocketFactory;
  19.     X509TrustManager mTrustManager;
  20.     /**
  21.      * 重连间隔时间
  22.      */
  23.     long mReconnectInterval;
  24.     /**
  25.      * 重连间隔时间的单位
  26.      */
  27.     TimeUnit mReconnectIntervalTimeUnit;
  28.  
  29.     public RxWebSocketBuilder(Context context) {
  30.         this.mContext = context.getApplicationContext();
  31.     }
  32.  
  33.     public RxWebSocketBuilder isPrintLog(boolean isPrintLog) {
  34.         this.mIsPrintLog = isPrintLog;
  35.         return this;
  36.     }
  37.  
  38.     public RxWebSocketBuilder logger(Logger.LogDelegate logDelegate) {
  39.         Logger.setDelegate(logDelegate);
  40.         return this;
  41.     }
  42.  
  43.     public RxWebSocketBuilder client(OkHttpClient client) {
  44.         this.mClient = client;
  45.         return this;
  46.     }
  47.  
  48.     public RxWebSocketBuilder sslSocketFactory(SSLSocketFactory sslSocketFactory, X509TrustManager trustManager) {
  49.         this.mSslSocketFactory = sslSocketFactory;
  50.         this.mTrustManager = trustManager;
  51.         return this;
  52.     }
  53.  
  54.     public RxWebSocketBuilder reconnectInterval(long reconnectInterval, TimeUnit reconnectIntervalTimeUnit) {
  55.         this.mReconnectInterval = reconnectInterval;
  56.         this.mReconnectIntervalTimeUnit = reconnectIntervalTimeUnit;
  57.         return this;
  58.     }
  59.  
  60.     public RxWebSocket build() {
  61.         return new RxWebSocket(this);
  62.     }
  63. }

3、基础缓存池

  1. public abstract class BaseCachePool<T extends ICacheTarget<T>> implements ICachePool<T>, Comparator<CacheItem<T>> {
  2.     /**
  3.      * 缓存池
  4.      */
  5.     private ConcurrentHashMap<String, LinkedList<CacheItem<T>>> mPool;
  7.     public BaseCachePool() {
  8.         mPool = new ConcurrentHashMap<>(8);
  9.     }
  10.  
  11.     @Override
  12.     public T obtain(String cacheKey) {
  13.         //缓存链
  14.         LinkedList<CacheItem<T>> cacheChain;
  15.         //没有缓存过，进行缓存
  16.         if (!mPool.containsKey(cacheKey)) {
  17.             cacheChain = new LinkedList<>();
  18.         } else {
  19.             cacheChain = mPool.get(cacheKey);
  20.         }
  21.         if (cacheChain == null) {
  22.             throw new NullPointerException("cacheChain 缓存链创建失败");
  23.         }
  24.         //未满最大缓存数量，生成一个实例
  25.         if (cacheChain.size() < onSetupMaxCacheCount()) {
  26.             T cache = onCreateCache();
  27.             CacheItem<T> cacheItem = new CacheItem<>(cache, System.currentTimeMillis());
  28.             cacheChain.add(cacheItem);
  29.             mPool.put(cacheKey, cacheChain);
  30.             return cache;
  31.         }
  32.         //达到最大缓存数量。按最近的使用时间排序，最近使用的放后面，每次取只取最前面（最久没有使用的）
  33.         Collections.sort(cacheChain, this);
  34.         CacheItem<T> cacheItem = cacheChain.getFirst();
  35.         cacheItem.setRecentlyUsedTime(System.currentTimeMillis());
  36.         //重置所有属性
  37.         T cacheTarget = cacheItem.getCacheTarget();
  38.         cacheTarget = onObtainCacheAfter(cacheTarget);
  39.         return cacheTarget;
  40.     }
  41.     
  42.     @Override
  43.     public T onObtainCacheAfter(ICacheTarget<T> cacheTarget) {
  44.         //默认调用reset方法进行重置，如果有其他需求，子类再进行复写
  45.         return cacheTarget.reset();
  46.     }
  47.  
  48.     @Override
  49.     public int compare(CacheItem<T> o1, CacheItem<T> o2) {
  50.         return Long.compare(o1.getRecentlyUsedTime(), o2.getRecentlyUsedTime());
  51.     }
  52. }

缓存模型

  1. public class WebSocketInfo implements Serializable, ICacheTarget<WebSocketInfo> {
  2.     private static final long serialVersionUID = -880481254453932113L;
  3.  
  4.     private WebSocket mWebSocket;
  5.     private String mStringMsg;
  6.     private ByteString mByteStringMsg;
  7.     /**
  8.      * 连接成功
  9.      */
  10.     private boolean isConnect;
  11.     /**
  12.      * 重连成功
  13.      */
  14.     private boolean isReconnect;
  15.     /**
  16.      * 准备重连
  17.      */
  18.     private boolean isPrepareReconnect;
  19.  
  20.     /**
  21.      * 重置
  22.      */
  23.     @Override
  24.     public WebSocketInfo reset() {
  25.         this.mWebSocket = null;
  26.         this.mStringMsg = null;
  27.         this.mByteStringMsg = null;
  28.         this.isConnect = false;
  29.         this.isReconnect = false;
  30.         this.isPrepareReconnect = false;
  31.         return this;
  32.     }
  33.  
  34.      //省略get、set方法
  35. }

4、具体实现

  • 将连接WebSocket封装到Observable的订阅回调中。
  • 以Map缓存Url和数据源，多个Url共享同一个连接对象。
  • 使用share操作符，让多个观察者同时订阅一个数据源。所有订阅者都取消订阅时，再断开连接。
  1. public class WebSocketWorkerImpl implements WebSocketWorker {
  2.     private static final String TAG = WebSocketWorkerImpl.class.getName();
  3.  
  4.     /**
  5.      * 上下文
  6.      */
  7.     private Context mContext;
  8.     /**
  9.      * 支持外部传入OkHttpClient
  10.      */
  11.     private OkHttpClient mClient;
  12.     /**
  13.      * 重连间隔时间
  14.      */
  15.     private long mReconnectInterval;
  16.     /**
  17.      * 重连间隔时间的单位
  18.      */
  19.     private TimeUnit mReconnectIntervalTimeUnit;
  20.  
  21.     /**
  22.      * 缓存观察者对象，Url对应一个Observable
  23.      */
  24.     private Map<String, Observable<WebSocketInfo>> mObservableCacheMap;
  25.     /**
  26.      * 缓存Url和对应的WebSocket实例，同一个Url共享一个WebSocket连接
  27.      */
  28.     private Map<String, WebSocket> mWebSocketPool;
  29.     /**
  30.      * WebSocketInfo缓存池
  31.      */
  32.     private final WebSocketInfoPool mWebSocketInfoPool;
  33.  
  34.     public WebSocketWorkerImpl(
  35.             Context context,
  36.             boolean isPrintLog,
  37.             Logger.LogDelegate logDelegate,
  38.             OkHttpClient client,
  39.             SSLSocketFactory sslSocketFactory,
  40.             X509TrustManager trustManager,
  41.             long reconnectInterval,
  42.             TimeUnit reconnectIntervalTimeUnit) {
  43.         this.mContext = context;
  44.         //配置Logger
  45.         Logger.setDelegate(logDelegate);
  46.         Logger.setLogPrintEnable(isPrintLog);
  47.         this.mClient = client;
  48.         //重试时间配置
  49.         this.mReconnectInterval = reconnectInterval;
  50.         this.mReconnectIntervalTimeUnit = reconnectIntervalTimeUnit;
  51.         //配置SSL
  52.         if (sslSocketFactory != null && trustManager != null) {
  53.             mClient = mClient.newBuilder().sslSocketFactory(sslSocketFactory, trustManager).build();
  54.         }
  55.         this.mObservableCacheMap = new HashMap<>(16);
  56.         this.mWebSocketPool = new HashMap<>(16);
  57.         mWebSocketInfoPool = new WebSocketInfoPool();
  58.     }
  59.  
  60.     @Override
  61.     public Observable<WebSocketInfo> get(String url) {
  62.         return getWebSocketInfo(url);
  63.     }
  64.  
  65.     @Override
  66.     public Observable<WebSocketInfo> get(String url, long timeout, TimeUnit timeUnit) {
  67.         return getWebSocketInfo(url, timeout, timeUnit);
  68.     }
  69.  
  70.     @Override
  71.     public Observable<Boolean> send(String url, String msg) {
  72.         return Observable.create(new ObservableOnSubscribe<Boolean>() {
  73.             @Override
  74.             public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
  75.                 WebSocket webSocket = mWebSocketPool.get(url);
  76.                 if (webSocket == null) {
  77.                     emitter.onError(new IllegalStateException("The WebSocket not open"));
  78.                 } else {
  79.                     emitter.onNext(webSocket.send(msg));
  80.                 }
  81.             }
  82.         });
  83.     }
  84.  
  85.     @Override
  86.     public Observable<Boolean> send(String url, ByteString byteString) {
  87.         return Observable.create(new ObservableOnSubscribe<Boolean>() {
  88.             @Override
  89.             public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
  90.                 WebSocket webSocket = mWebSocketPool.get(url);
  91.                 if (webSocket == null) {
  92.                     emitter.onError(new IllegalStateException("The WebSocket not open"));
  93.                 } else {
  94.                     emitter.onNext(webSocket.send(byteString));
  95.                 }
  96.             }
  97.         });
  98.     }
  99.  
  100.     @Override
  101.     public Observable<Boolean> asyncSend(String url, String msg) {
  102.         return getWebSocket(url)
  103.                 .take(1)
  104.                 .map(new Function<WebSocket, Boolean>() {
  105.                     @Override
  106.                     public Boolean apply(WebSocket webSocket) throws Exception {
  107.                         return webSocket.send(msg);
  108.                     }
  109.                 });
  110.     }
  111.  
  112.     @Override
  113.     public Observable<Boolean> asyncSend(String url, ByteString byteString) {
  114.         return getWebSocket(url)
  115.                 .take(1)
  116.                 .map(new Function<WebSocket, Boolean>() {
  117.                     @Override
  118.                     public Boolean apply(WebSocket webSocket) throws Exception {
  119.                         return webSocket.send(byteString);
  120.                     }
  121.                 });
  122.     }
  123.  
  124.     @Override
  125.     public Observable<Boolean> close(String url) {
  126.         return Observable.create(new ObservableOnSubscribe<WebSocket>() {
  127.             @Override
  128.             public void subscribe(ObservableEmitter<WebSocket> emitter) throws Exception {
  129.                 WebSocket webSocket = mWebSocketPool.get(url);
  130.                 if (webSocket == null) {
  131.                     emitter.onError(new NullPointerException("url:" + url + " WebSocket must be not null"));
  132.                 } else {
  133.                     emitter.onNext(webSocket);
  134.                 }
  135.             }
  136.         }).map(new Function<WebSocket, Boolean>() {
  137.             @Override
  138.             public Boolean apply(WebSocket webSocket) throws Exception {
  139.                 return closeWebSocket(webSocket);
  140.             }
  141.         });
  142.     }
  143.  
  144.     @Override
  145.     public boolean closeNow(String url) {
  146.         return closeWebSocket(mWebSocketPool.get(url));
  147.     }
  148.  
  149.     @Override
  150.     public Observable<List<Boolean>> closeAll() {
  151.         return Observable
  152.                 .just(mWebSocketPool)
  153.                 .map(new Function<Map<String, WebSocket>, Collection<WebSocket>>() {
  154.                     @Override
  155.                     public Collection<WebSocket> apply(Map<String, WebSocket> webSocketMap) throws Exception {
  156.                         return webSocketMap.values();
  157.                     }
  158.                 })
  159.                 .concatMap(new Function<Collection<WebSocket>, ObservableSource<WebSocket>>() {
  160.                     @Override
  161.                     public ObservableSource<WebSocket> apply(Collection<WebSocket> webSockets) throws Exception {
  162.                         return Observable.fromIterable(webSockets);
  163.                     }
  164.                 }).map(new Function<WebSocket, Boolean>() {
  165.                     @Override
  166.                     public Boolean apply(WebSocket webSocket) throws Exception {
  167.                         return closeWebSocket(webSocket);
  168.                     }
  169.                 }).collect(new Callable<List<Boolean>>() {
  170.                     @Override
  171.                     public List<Boolean> call() throws Exception {
  172.                         return new ArrayList<>();
  173.                     }
  174.                 }, new BiConsumer<List<Boolean>, Boolean>() {
  175.                     @Override
  176.                     public void accept(List<Boolean> list, Boolean isCloseSuccess) throws Exception {
  177.                         list.add(isCloseSuccess);
  178.                     }
  179.                 }).toObservable();
  180.     }
  181.  
  182.     @Override
  183.     public void closeAllNow() {
  184.         for (Map.Entry<String, WebSocket> entry : mWebSocketPool.entrySet()) {
  185.             closeWebSocket(entry.getValue());
  186.         }
  187.     }
  188.  
  189.     /**
  190.      * 是否有连接
  191.      */
  192.     private boolean hasWebSocketConnection(String url) {
  193.         return mWebSocketPool.get(url) != null;
  194.     }
  195.  
  196.     /**
  197.      * 关闭WebSocket连接
  198.      */
  199.     private boolean closeWebSocket(WebSocket webSocket) {
  200.         if (webSocket == null) {
  201.             return false;
  202.         }
  203.         WebSocketCloseEnum normalCloseEnum = WebSocketCloseEnum.USER_EXIT;
  204.         boolean result = webSocket.close(normalCloseEnum.getCode(), normalCloseEnum.getReason());
  205.         if (result) {
  206.             removeUrlWebSocketMapping(webSocket);
  207.         }
  208.         return result;
  209.     }
  210.  
  211.     /**
  212.      * 移除Url和WebSocket的映射
  213.      */
  214.     private void removeUrlWebSocketMapping(WebSocket webSocket) {
  215.         for (Map.Entry<String, WebSocket> entry : mWebSocketPool.entrySet()) {
  216.             if (entry.getValue() == webSocket) {
  217.                 String url = entry.getKey();
  218.                 mObservableCacheMap.remove(url);
  219.                 mWebSocketPool.remove(url);
  220.             }
  221.         }
  222.     }
  223.  
  224.     private void removeWebSocketCache(WebSocket webSocket) {
  225.         for (Map.Entry<String, WebSocket> entry : mWebSocketPool.entrySet()) {
  226.             if (entry.getValue() == webSocket) {
  227.                 String url = entry.getKey();
  228.                 mWebSocketPool.remove(url);
  229.             }
  230.         }
  231.     }
  232.  
  233.     public Observable<WebSocket> getWebSocket(String url) {
  234.         return getWebSocketInfo(url)
  235.                 .filter(new Predicate<WebSocketInfo>() {
  236.                     @Override
  237.                     public boolean test(WebSocketInfo webSocketInfo) throws Exception {
  238.                         return webSocketInfo.getWebSocket() != null;
  239.                     }
  240.                 })
  241.                 .map(new Function<WebSocketInfo, WebSocket>() {
  242.                     @Override
  243.                     public WebSocket apply(WebSocketInfo webSocketInfo) throws Exception {
  244.                         return webSocketInfo.getWebSocket();
  245.                     }
  246.                 });
  247.     }
  248.  
  249.     public Observable<WebSocketInfo> getWebSocketInfo(String url) {
  250.         return getWebSocketInfo(url, 5, TimeUnit.SECONDS);
  251.     }
  252.  
  253.     public synchronized Observable<WebSocketInfo> getWebSocketInfo(final String url, final long timeout, final TimeUnit timeUnit) {
  254.         //先从缓存中取
  255.         Observable<WebSocketInfo> observable = mObservableCacheMap.get(url);
  256.         if (observable == null) {
  257.             //缓存中没有，新建
  258.             observable = Observable
  259.                     .create(new WebSocketOnSubscribe(url))
  260.                     .retry()
  261.                     //因为有share操作符，只有当所有观察者取消注册时，这里才会回调
  262.                     .doOnDispose(new Action() {
  263.                         @Override
  264.                         public void run() throws Exception {
  265.                             //所有都不注册了，关闭连接
  266.                             closeNow(url);
  267.                             Logger.d(TAG, "所有观察者都取消注册，关闭连接...");
  268.                         }
  269.                     })
  270.                     //Share操作符，实现多个观察者对应一个数据源
  271.                     .share()
  272.                     //将回调都放置到主线程回调，外部调用方直接观察，实现响应回调方法做UI处理
  273.                     .subscribeOn(Schedulers.io())
  274.                     .observeOn(AndroidSchedulers.mainThread());
  275.             //将数据源缓存
  276.             mObservableCacheMap.put(url, observable);
  277.         } else {
  278.             //缓存中有，从连接池中取出
  279.             WebSocket webSocket = mWebSocketPool.get(url);
  280.             if (webSocket != null) {
  281.                 observable = observable.startWith(createConnect(url, webSocket));
  282.             }
  283.         }
  284.         return observable;
  285.     }
  286.  
  287.     /**
  288.      * 组装数据源
  289.      */
  290.     private final class WebSocketOnSubscribe implements ObservableOnSubscribe<WebSocketInfo> {
  291.         private String mWebSocketUrl;
  292.         private WebSocket mWebSocket;
  293.         private boolean isReconnecting = false;
  294.  
  295.         public WebSocketOnSubscribe(String webSocketUrl) {
  296.             this.mWebSocketUrl = webSocketUrl;
  297.         }
  298.  
  299.         @Override
  300.         public void subscribe(ObservableEmitter<WebSocketInfo> emitter) throws Exception {
  301.             //因为retry重连不能设置延时，所以只能这里延时，降低发送频率
  302.             if (mWebSocket == null && isReconnecting) {
  303.                 if (Thread.currentThread() != Looper.getMainLooper().getThread()) {
  304.                     long millis = mReconnectIntervalTimeUnit.toMillis(mReconnectInterval);
  305.                     if (millis == 0) {
  306.                         millis = 1000;
  307.                     }
  308.                     SystemClock.sleep(millis);
  309.                 }
  310.             }
  311.             initWebSocket(emitter);
  312.         }
  313.  
  314.         private Request createRequest(String url) {
  315.             return new Request.Builder().get().url(url).build();
  316.         }
  317.  
  318.         /**
  319.          * 初始化WebSocket
  320.          */
  321.         private synchronized void initWebSocket(ObservableEmitter<WebSocketInfo> emitter) {
  322.             if (mWebSocket == null) {
  323.                 mWebSocket = mClient.newWebSocket(createRequest(mWebSocketUrl), new WebSocketListener() {
  324.                     @Override
  325.                     public void onOpen(WebSocket webSocket, Response response) {
  326.                         super.onOpen(webSocket, response);
  327.                         //连接成功
  328.                         if (!emitter.isDisposed()) {
  329.                             mWebSocketPool.put(mWebSocketUrl, mWebSocket);
  330.                             //重连成功
  331.                             if (isReconnecting) {
  332.                                 emitter.onNext(createReconnect(mWebSocketUrl, webSocket));
  333.                             } else {
  334.                                 emitter.onNext(createConnect(mWebSocketUrl, webSocket));
  335.                             }
  336.                         }
  337.                         isReconnecting = false;
  338.                     }
  339.  
  340.                     @Override
  341.                     public void onMessage(WebSocket webSocket, String text) {
  342.                         super.onMessage(webSocket, text);
  343.                         //收到消息
  344.                         if (!emitter.isDisposed()) {
  345.                             emitter.onNext(createReceiveStringMsg(mWebSocketUrl, webSocket, text));
  346.                         }
  347.                     }
  348.  
  349.                     @Override
  350.                     public void onMessage(WebSocket webSocket, ByteString bytes) {
  351.                         super.onMessage(webSocket, bytes);
  352.                         //收到消息
  353.                         if (!emitter.isDisposed()) {
  354.                             emitter.onNext(createReceiveByteStringMsg(mWebSocketUrl, webSocket, bytes));
  355.                         }
  356.                     }
  357.  
  358.                     @Override
  359.                     public void onClosed(WebSocket webSocket, int code, String reason) {
  360.                         super.onClosed(webSocket, code, reason);
  361.                         if (!emitter.isDisposed()) {
  362.                             emitter.onNext(createClose(mWebSocketUrl));
  363.                         }
  364.                     }
  365.  
  366.                     @Override
  367.                     public void onFailure(WebSocket webSocket, Throwable throwable, Response response) {
  368.                         super.onFailure(webSocket, throwable, response);
  369.                         isReconnecting = true;
  370.                         mWebSocket = null;
  371.                         //移除WebSocket缓存，retry重试重新连接
  372.                         removeWebSocketCache(webSocket);
  373.                         if (!emitter.isDisposed()) {
  374.                             emitter.onNext(createPrepareReconnect(mWebSocketUrl));
  375.                             //失败发送onError，让retry操作符重试
  376.                             emitter.onError(new ImproperCloseException());
  377.                         }
  378.                     }
  379.                 });
  380.             }
  381.         }
  382.     }
  383.  
  384.  
  385.     private WebSocketInfo createConnect(String url, WebSocket webSocket) {
  386.         return mWebSocketInfoPool.obtain(url)
  387.                 .setWebSocket(webSocket)
  388.                 .setConnect(true);
  389.     }
  390.  
  391.     private WebSocketInfo createReconnect(String url, WebSocket webSocket) {
  392.         return mWebSocketInfoPool.obtain(url)
  393.                 .setWebSocket(webSocket)
  394.                 .setReconnect(true);
  395.     }
  396.  
  397.     private WebSocketInfo createPrepareReconnect(String url) {
  398.         return mWebSocketInfoPool.obtain(url)
  399.                 .setPrepareReconnect(true);
  400.     }
  401.  
  402.     private WebSocketInfo createReceiveStringMsg(String url, WebSocket webSocket, String stringMsg) {
  403.         return mWebSocketInfoPool.obtain(url)
  404.                 .setConnect(true)
  405.                 .setWebSocket(webSocket)
  406.                 .setStringMsg(stringMsg);
  407.     }
  408.  
  409.     private WebSocketInfo createReceiveByteStringMsg(String url, WebSocket webSocket, ByteString byteMsg) {
  410.         return mWebSocketInfoPool.obtain(url)
  411.                 .setConnect(true)
  412.                 .setWebSocket(webSocket)
  413.                 .setByteStringMsg(byteMsg);
  414.     }
  415.  
  416.     private WebSocketInfo createClose(String url) {
  417.         return mWebSocketInfoPool.obtain(url);
  418.     }
  419. }

定时发送心跳维持连接

因为WebSocket断线后，后端不能马上知道连接已经断开，所以需要一个心跳消息保持双方通信。

实现心跳，本质就是一个定时消息，我们使用RxJava的interval操作符定时执行任务，这里我的消息需要增加一个时间戳，所以我加上了timestamp操作符来给每一次执行结果附加一个时间戳。

  • 心跳信息json的生成，我们期望外部进行生成，例如Gson序列化为Json，或者FastJson处理，或者增加其他通用参数等，不应该在WebSocket基础库中写，所以提供了一个HeartBeatGenerateCallback回调进行生成Json

1.网络没有开启时，不发送心跳消息

  1. public class NetworkUtil {
  2.     private NetworkUtil() {
  3.     }
  4.  
  5.     /**
  6.      * 当前是否有网络状态
  7.      *
  8.      * @param context  上下文
  9.      * @param needWifi 是否需要wifi网络
  10.      */
  11.     public static boolean hasNetWorkStatus(Context context, boolean needWifi) {
  12.         NetworkInfo info = getActiveNetwork(context);
  13.         if (info == null) {
  14.             return false;
  15.         }
  16.         if (!needWifi) {
  17.             return info.isAvailable();
  18.         } else if (info.getType() == ConnectivityManager.TYPE_WIFI) {
  19.             return info.isAvailable();
  20.         }
  21.         return false;
  22.     }
  23.  
  24.     /**
  25.      * 获取活动网络连接信息
  26.      *
  27.      * @param context 上下文
  28.      * @return NetworkInfo
  29.      */
  30.     public static NetworkInfo getActiveNetwork(Context context) {
  31.         ConnectivityManager mConnMgr = (ConnectivityManager) context
  32.                 .getSystemService(Context.CONNECTIVITY_SERVICE);
  33.         if (mConnMgr == null) {
  34.             return null;
  35.         }
  36.         // 获取活动网络连接信息
  37.         return mConnMgr.getActiveNetworkInfo();
  38.     }
  39. }

2.心跳回调接口，让外部生成心跳json

  1. public interface HeartBeatGenerateCallback {
  2.     /**
  3.      * 当需要生成心跳信息时回调
  4.      *
  5.      * @param timestamp 当前时间戳
  6.      * @return 要发送的心跳信息
  7.      */
  8.     String onGenerateHeartBeatMsg(long timestamp);
  9. }

3.发送心跳消息，需要制定Url地址、间隔时间，间隔时间单位，心跳消息生成回调

  1. @Override
  2. public Observable<Boolean> heartBeat(String url, int period, TimeUnit unit,
  3.                                      HeartBeatGenerateCallback heartBeatGenerateCallback) {
  4. if (heartBeatGenerateCallback == null) {
  5.     return Observable.error(new NullPointerException("heartBeatGenerateCallback == null"));
  6. }
  7. return Observable
  8.         .interval(period, unit)
  9.         //timestamp操作符，给每个事件加一个时间戳
  10.         .timestamp()
  11.         .retry()
  12.         .flatMap(new Function<Timed<Long>, ObservableSource<Boolean>>() {
  13.             @Override
  14.             public ObservableSource<Boolean> apply(Timed<Long> timed) throws Exception {
  15.                 long timestamp = timed.time();
  16.                 //判断网络，存在网络才发消息，否则直接返回发送心跳失败
  17.                 if (mContext != null && NetworkUtil.hasNetWorkStatus(mContext, false)) {
  18.                     String heartBeatMsg = heartBeatGenerateCallback.onGenerateHeartBeatMsg(timestamp);
  19.                     Logger.d(TAG, "发送心跳消息: " + heartBeatMsg);
  20.                     if (hasWebSocketConnection(url)) {
  21.                         return send(url, heartBeatMsg);
  22.                     } else {
  23.                         //这里必须用异步发送，如果切断网络，再重连，缓存的WebSocket会被清除，此时再重连网络
  24.                         //是没有WebSocket连接可用的，所以就需要异步连接完成后，再发送
  25.                         return asyncSend(url, heartBeatMsg);
  26.                     }
  27.                 } else {
  28.                     Logger.d(TAG, "无网络连接，不发送心跳，下次网络连通时，再次发送心跳");
  29.                     return Observable.create(new ObservableOnSubscribe<Boolean>() {
  30.                         @Override
  31.                         public void subscribe(ObservableEmitter<Boolean> emitter) throws Exception {
  32.                             emitter.onNext(false);
  33.                         }
  34.                     });
  35.                 }
  36.             }
  37.         });
  38. }

实现重连

重连配置RxJava，有个天然优势就是RxJava提供了Retry操作符，支持重试，我们在onFailure()连接失败回调中手动发出onError()，让数据源增加retry操作符进行重试，就会重新走到数据源的订阅回调重新连接WebSocket

  1. private final class WebSocketOnSubscribe implements ObservableOnSubscribe<WebSocketInfo> {
  2.         private String mWebSocketUrl;
  3.         private WebSocket mWebSocket;
  4.         private boolean isReconnecting = false;
  5.  
  6.         public WebSocketOnSubscribe(String webSocketUrl) {
  7.             this.mWebSocketUrl = webSocketUrl;
  8.         }
  9.  
  10.         @Override
  11.         public void subscribe(ObservableEmitter<WebSocketInfo> emitter) throws Exception {
  12.             //...
  13.         }
  14.  
  15.         private Request createRequest(String url) {
  16.             return new Request.Builder().get().url(url).build();
  17.         }
  18.  
  19.         /**
  20.          * 初始化WebSocket
  21.          */
  22.         private synchronized void initWebSocket(ObservableEmitter<WebSocketInfo> emitter) {
  23.             if (mWebSocket == null) {
  24.                 mWebSocket = mClient.newWebSocket(createRequest(mWebSocketUrl), new WebSocketListener() {
  25.                     @Override
  26.                     public void onOpen(WebSocket webSocket, Response response) {
  27.                         super.onOpen(webSocket, response);
  28.                         //连接成功
  29.                         if (!emitter.isDisposed()) {
  30.                             mWebSocketPool.put(mWebSocketUrl, mWebSocket);
  31.                             //重连成功
  32.                             if (isReconnecting) {
  33.                                 emitter.onNext(createReconnect(mWebSocketUrl, webSocket));
  34.                             } else {
  35.                                 emitter.onNext(createConnect(mWebSocketUrl, webSocket));
  36.                             }
  37.                         }
  38.                         isReconnecting = false;
  39.                     }
  40.  
  41.                     @Override
  42.                     public void onMessage(WebSocket webSocket, String text) {
  43.                         super.onMessage(webSocket, text);
  44.                         //收到消息
  45.                         if (!emitter.isDisposed()) {
  46.                             emitter.onNext(createReceiveStringMsg(mWebSocketUrl, webSocket, text));
  47.                         }
  48.                     }
  49.  
  50.                     @Override
  51.                     public void onMessage(WebSocket webSocket, ByteString bytes) {
  52.                         super.onMessage(webSocket, bytes);
  53.                         //收到消息
  54.                         if (!emitter.isDisposed()) {
  55.                             emitter.onNext(createReceiveByteStringMsg(mWebSocketUrl, webSocket, bytes));
  56.                         }
  57.                     }
  58.  
  59.                     @Override
  60.                     public void onClosed(WebSocket webSocket, int code, String reason) {
  61.                         super.onClosed(webSocket, code, reason);
  62.                         if (!emitter.isDisposed()) {
  63.                             emitter.onNext(createClose(mWebSocketUrl));
  64.                         }
  65.                     }
  66.  
  67.                     @Override
  68.                     public void onFailure(WebSocket webSocket, Throwable throwable, Response response) {
  69.                         super.onFailure(webSocket, throwable, response);
  70.                         isReconnecting = true;
  71.                         mWebSocket = null;
  72.                         //移除WebSocket缓存，retry重试重新连接
  73.                         removeWebSocketCache(webSocket);
  74.                         if (!emitter.isDisposed()) {
  75.                             emitter.onNext(createPrepareReconnect(mWebSocketUrl));
  76.                             //失败发送onError，让retry操作符重试
  77.                             emitter.onError(new ImproperCloseException());
  78.                         }
  79.                     }
  80.                 });
  81.             }
  82.         }
  83.     }

以上，基于Rxjava对WebSocket的封装基本完成，以下是项目中的实际使用

使用

  • 使用RxWebSocketBuilder，构建RxWebSocket
    1. //自定义OkHttpClient
    2. OkHttpClient mClient = new OkHttpClient.Builder()
    3.         .readTimeout(3, TimeUnit.SECONDS)//设置读取超时时间
    4.         .writeTimeout(3, TimeUnit.SECONDS)//设置写的超时时间
    5.         .connectTimeout(3, TimeUnit.SECONDS)//设置连接超时时间
    6.         .build();
    7.  
    8. //RxWebSocketBuilder构建RxWebSocket
    9. RxWebSocket rxWebSocket = new RxWebSocketBuilder(context)
    10.                 //是否打印Log
    11.                 .isPrintLog(true)
    12.                 //5秒无响应则重连
    13.                 .reconnectInterval(5, TimeUnit.SECONDS)
    14.                 .client(mClient)
    15.                 .build();

  • 连接Url地址
    1. String url = "ws://xxxxxxxxx"
    2. //开始连接
    3. rxWebSocket.get(url)
    4.     //切换到子线程去连接
    5.     .compose(RxSchedulerUtil.ioToMain())
    6.     //绑定生命周期
    7.     .as(RxLifecycleUtil.bindLifecycle(mLifecycleOwner))
    8.     .subscribe(new Consumer<WebSocketInfo>() {
    9.         @Override
    10.         public void accept(WebSocketInfo webSocketInfo) throws Exception {
    11.             String json = webSocketInfo.getStringMsg();
    12.             //业务层的json解析
    13.             ...
    14.         }
    15.     });

  • 同步发送消息（必须确保连接正常，否则发送失败）
    1. rxWebSocket.send(url, "我是消息")
    2.     .compose(RxSchedulerUtil.ioToMain())
    3.     .as(RxLifecycleUtil.bindLifecycle(mLifecycleOwner))
    4.     .subscribe(new Consumer<Boolean>() {
    5.             @Override
    6.             public void accept(Boolean isSuccess) throws Exception {
    7.                 if(isSuccess) {
    8.                       //发送成功
    9.                 } ele {
    10.                       //发送失败
    11.                 }
    12.             }
    13.     });

  • 异步发送消息（不需要确保连接正常，如果未连接会连接成功后自动发送）
    1. rxWebSocket.asyncSend(url, "我是消息")
    2.     .compose(RxSchedulerUtil.ioToMain())
    3.     .as(RxLifecycleUtil.bindLifecycle(mLifecycleOwner))
    4.     .subscribe(new Consumer<Boolean>() {
    5.             @Override
    6.             public void accept(Boolean isSuccess) throws Exception {
    7.                 if(isSuccess) {
    8.                       //发送成功
    9.                 } ele {
    10.                       //发送失败
    11.                 }
    12.             }
    13.     });

  • 发送心跳包
    1. rxWebSocket.heartBeat(url, 6 ,TimeUnit.SECONDS, new HeartBeatGenerateCallback() {
    2.     @Override
    3.     public String onGenerateHeartBeatMsg(long timestamp) {
    4.         //生成心跳Json，业务模块处理，例如后端需要秒值，我们除以1000换算为秒。
    5.         //后续可以在这里配置通用参数等
    6.         return GsonUtil.toJson(new HeartBeatMsgRequestModel(WssCommandTypeEnum.HEART_BEAT.getCode(),
    7.                 String.valueOf(timestamp / 1000)));
    8.     }
    9. });

总结 

Okhttp的WebSocket使用比较简单，基本都是发起请求和配置回调2个步骤，再使用send()方法发送消息，简单使用可以不需要做封装，Java使用可以基于Rxjava做封装，而Kotlin使用可以用协程做封装，后续有时间我可能会填这个坑

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号