RxAndroidBLE已连接状态下无法向IoT设备发送数据及实现300ms定时发送的问题排查

一、为什么会出现「已连接却发送失败」的矛盾情况？ #

你遇到的BleAlreadyConnectedException和isConnected()返回false的矛盾，核心原因是没有正确复用BLE连接Observable，加上状态判断的时机/方式问题：

1. 连接Observable未被共享
   你注释掉了compose(ReplayingShare.instance())，这导致每次订阅connectionObservable（比如发送数据时调用.firstOrError()）都会触发rxBleDevice.establishConnection(false)执行一次新的连接尝试。但实际上，第一次点击连接后，连接已经建立（因为takeUntil(disconnectTriggerSubject)还没触发断开），此时再次尝试建立连接就会抛出「已连接」的异常。

2. getConnectionState()的状态更新延迟
   rxBleDevice.getConnectionState()的状态同步可能存在短暂延迟，或者因为连接Observable没有被共享，内部状态和外部API的状态不一致，导致你调用isConnected()时拿到的是旧状态。

修复连接复用问题 #

你之前遇到的类型推断错误，是因为Java的泛型推断需要明确类型，只需要给ReplayingShare.instance()加上显式的类型转换即可：

connectionObservable = rxBleDevice.establishConnection(false)
        .takeUntil(disconnectTriggerSubject)
        .compose((ObservableTransformer<RxBleConnection, RxBleConnection>) ReplayingShare.instance());

ReplayingShare的作用是让所有订阅者共享同一个连接Observable，避免重复发起连接，同时会缓存最新的连接状态，新订阅者可以直接拿到已建立的连接。

二、如何实现每300ms向IoT设备发送一次数据？ #

要实现定时发送，我们可以结合Observable.interval来定时触发，同时复用已共享的连接Observable，确保每次发送都使用同一个活跃连接：

完整实现步骤 #

1. 确保连接Observable已经通过ReplayingShare共享（上面的修复已完成）
2. 创建定时发送的Observable，绑定到连接生命周期
3. 管理订阅的Disposable，避免内存泄漏

示例代码：

// 定义定时发送的Disposable（作为成员变量）
private Disposable periodicSendDisposable;

// 启动定时发送的方法
private void startPeriodicDataSend(String data, UUID characteristicUuid) {
    // 先取消之前的定时发送（如果存在）
    if (periodicSendDisposable != null && !periodicSendDisposable.isDisposed()) {
        periodicSendDisposable.dispose();
    }

    periodicSendDisposable = Observable.interval(300, TimeUnit.MILLISECONDS)
            // 切换到Android主线程（如果需要更新UI）
            .observeOn(AndroidSchedulers.mainThread())
            // 每次定时触发时，获取已建立的连接
            .flatMapSingle(tick -> connectionObservable
                    .firstOrError()
                    .flatMap(connection -> connection.writeCharacteristic(characteristicUuid, HexString.hexToBytes(data))))
            .subscribe(
                    bytes -> onWriteSuccess(bytes),
                    throwable -> {
                        onWriteFailure(throwable);
                        // 如果发送失败，可以选择停止定时任务，或者继续尝试
                        if (!periodicSendDisposable.isDisposed()) {
                            periodicSendDisposable.dispose();
                        }
                    }
            );

    // 添加到CompositeDisposable统一管理
    compositeDisposable.add(periodicSendDisposable);
}

// 在断开连接时，记得取消定时发送
private void triggerDisconnect() {
    disconnectTriggerSubject.onNext(Unit.INSTANCE);
    if (periodicSendDisposable != null && !periodicSendDisposable.isDisposed()) {
        periodicSendDisposable.dispose();
    }
    rxBleDevice = null;
}

注意事项 #

• 连接状态的判断：现在你可以通过connectionObservable的状态来判断是否连接，比如：

  private boolean isConnected() {
      return !connectionObservable.isEmpty().blockingGet();
  }
  

  或者监听connectionObservable的订阅状态，更准确地获取连接状态。

• 数据发送的可靠性：BLE的write操作可能存在丢包，你可以根据需求添加重试逻辑，或者使用writeCharacteristic的重载方法（带确认的write）。

• 内存泄漏：所有的Disposable都要添加到CompositeDisposable中，在Fragment/Activity销毁时调用compositeDisposable.dispose()。

内容的提问来源于stack exchange，提问作者Arkadiusz