基于RxAndroidBle与RxJava2实现BLE特征值多次读取的问题

嘿，我之前做BLE项目时也碰到过这种分段数据读取的需求，刚好能用RxJava的操作符完美解决，结合RxAndroidBle的API给你几个实用的方案：

1. 固定次数重复读取 #

如果已经明确知道要读取的次数，比如要拉取3段数据，可以把readCharacteristic()返回的Single转成Observable，再用repeat()指定重复次数：

// 假设你已经持有连接后的RxBleConnection实例和目标特征值characteristic
Observable<ByteArray> fixedCountRead = Observable.defer(() -> 
    rxBleConnection.readCharacteristic(characteristic).toObservable()
).repeat(3); // 读取3次

fixedCountRead.subscribe(
    segmentData -> {
        // 处理每一段读取到的数据
        handleSegment(segmentData);
    },
    error -> {
        // 处理读取失败的情况，比如连接断开、特征值不可读
        Log.e("BLE_READ", "读取特征值失败", error);
    },
    () -> {
        // 所有次数读取完成后的回调
        Log.d("BLE_READ", "固定次数读取完成");
    }
);

这里用Observable.defer()的原因是：每次重复读取都需要创建一个新的readCharacteristic()请求，不能复用同一个Single实例，否则只会执行一次读取操作。

2. 读取到结束标志后停止 #

如果BLE设备会在最后一段数据里带上结束标识（比如字节数组末尾是0x00，或者数据长度达到某个阈值），可以用repeatUntil()或takeWhile()来控制停止时机：

方案A：用repeatUntil()判断停止条件 #

// 用AtomicReference保存最后一次读取的结果，保证线程安全
AtomicReference<byte[]> lastSegment = new AtomicReference<>();

Observable<ByteArray> continuousRead = Observable.defer(() -> 
    rxBleConnection.readCharacteristic(characteristic).toObservable()
).doOnNext(lastSegment::set) // 每次读取后保存结果
.repeatUntil(() -> {
    byte[] data = lastSegment.get();
    // 判断是否是最后一段：比如最后一个字节为0x00
    return data != null && data.length > 0 && data[data.length - 1] == 0x00;
});

continuousRead.subscribe(
    segmentData -> handleSegment(segmentData),
    error -> Log.e("BLE_READ", "读取出错", error),
    () -> Log.d("BLE_READ", "所有分段数据读取完成")
);

方案B：用takeWhile()过滤继续读取的条件 #

Observable<ByteArray> continuousRead = Observable.defer(() -> 
    rxBleConnection.readCharacteristic(characteristic).toObservable()
).takeWhile(segmentData -> {
    // 返回true表示继续读取，false则停止
    return segmentData[segmentData.length - 1] != 0x00;
});

continuousRead.subscribe(
    segmentData -> handleSegment(segmentData),
    error -> Log.e("BLE_READ", "读取出错", error),
    () -> Log.d("BLE_READ", "分段读取完成")
);

3. 定时轮询读取 #

如果需要每隔一段时间读取一次（比如设备会周期性更新分段数据），可以用interval()来触发读取，再结合takeWhile()控制停止：

Observable.interval(1, TimeUnit.SECONDS) // 每隔1秒触发一次
    .concatMap(tick -> 
        // 用concatMap保证读取操作串行执行，避免BLE操作并发冲突
        rxBleConnection.readCharacteristic(characteristic).toObservable()
    )
    .takeWhile(segmentData -> {
        // 判断是否还需要继续轮询
        return !isAllDataReceived(segmentData);
    })
    .subscribe(
        segmentData -> handleSegment(segmentData),
        error -> Log.e("BLE_READ", "轮询读取失败", error),
        () -> Log.d("BLE_READ", "轮询读取完成")
    );

这里用concatMap而不是flatMap，是因为BLE操作不能并发执行，否则容易被系统拒绝，concatMap会等待上一次读取完成后再发起下一次请求。

几个重要的注意事项 #

• 保持连接稳定：在读取过程中要确保BLE连接没有断开，可以监听rxBleConnection.observeConnectionStateChanges()来处理连接断开的情况，及时停止读取操作。
• 重试失败请求：如果偶尔出现读取失败，可以用retry()操作符添加重试逻辑，比如retry(3)表示失败后重试3次。
• 避免过度频繁请求：BLE设备的操作队列有长度限制，过于频繁的读取请求会导致失败，建议在两次读取之间留适当间隔，或者用串行化的操作符（比如concatMap）控制请求节奏。

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