ReactiveNetwork 库时如何实现网络状态监听的

2021 年 12 月 08 日
本文字数：1401 字
阅读完需：约 5 分钟

1.知识点

简介：Android library listening network connection state and Internet connectivity with RxJava Observables。一个基于 RxJava 的监听网络连接和状态的 Android 库。
代码仓库：

git clone https://github.com/pwittchen/ReactiveNetwork.git

复制代码

使用方法：

Disposable disposable = ReactiveNetwork            .observeNetworkConnectivity(this) // 观察网络连接            .subscribeOn(Schedulers.io())            .observeOn(AndroidSchedulers.mainThread())            .subscribe(connectivity -> { // 网络状态变化时执行该回调                boolean connected = connectivity.available();                Log.e("cclin","networkchange isConnected = "+connected);            });

复制代码

备注：不会分析异步分发逻辑，主要侧重于分析如何兼容不同的系统版本的网络监听。

2.原理

ReactiveNetwork.java，使用一个策略设计模式，根据不同的系统版本，分发给不同的策略之类处理

 public static Observable<Connectivity> observeNetworkConnectivity(final Context context) {    final NetworkObservingStrategy strategy;
    if (Preconditions.isAtLeastAndroidMarshmallow()) {      strategy = new MarshmallowNetworkObservingStrategy();    } else if (Preconditions.isAtLeastAndroidLollipop()) {      strategy = new LollipopNetworkObservingStrategy();    } else {      strategy = new PreLollipopNetworkObservingStrategy();    }
    return observeNetworkConnectivity(context, strategy);  }

复制代码

MarshmallowNetworkObservingStrategy.java，查看 6.0 以后系统版本的网络连接监听

@Override public Observable<Connectivity> observeNetworkConnectivity(final Context context) {    final String service = Context.CONNECTIVITY_SERVICE;    final ConnectivityManager manager = (ConnectivityManager) context.getSystemService(service);    networkCallback = createNetworkCallback(context);
    registerIdleReceiver(context);
    final NetworkRequest request =        new NetworkRequest.Builder().addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)            .addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_RESTRICTED)            .build();
    manager.registerNetworkCallback(request, networkCallback);
    return connectivitySubject.toFlowable(BackpressureStrategy.LATEST).doOnCancel(new Action() {      @Override public void run() {        tryToUnregisterCallback(manager);        tryToUnregisterReceiver(context);      }    }).doAfterNext(new Consumer<Connectivity>() {      @Override      public void accept(final Connectivity connectivity) {        lastConnectivity = connectivity;      }    }).flatMap(new Function<Connectivity, Publisher<Connectivity>>() {      @Override      public Publisher<Connectivity> apply(final Connectivity connectivity) {        return propagateAnyConnectedState(lastConnectivity, connectivity);      }    }).startWith(Connectivity.create(context)).distinctUntilChanged().toObservable();  }