22、蓝牙与协程/Flow：Kotlin协程封装蓝牙回调、Flow实现扫描流、Room存储蓝牙设备信息

说实话，蓝牙开发最让人头疼的是什么？

不是协议有多复杂，而是回调地狱。你想想看，一个简单的设备扫描，要注册广播接收器，要处理各种状态回调，还要手动管理生命周期。代码写出来跟千层饼似的，一层套一层。

我个人习惯用协程来改造这一切。Kotlin协程配合Flow，能把蓝牙那些乱七八糟的回调，变成一条干净的数据流。今天我们就来聊聊怎么做到这一点。

为什么蓝牙回调需要协程改造？

先看一个典型的蓝牙扫描代码：

val scanCallback = object : ScanCallback() {
    override fun onScanResult(callbackType: Int, result: ScanResult?) {
        // 处理扫描结果
    }
    override fun onBatchScanResults(results: MutableList<ScanResult>?) {
        // 批量结果
    }
    override fun onScanFailed(errorCode: Int) {
        // 失败处理
    }
}
bluetoothLeScanner.startScan(scanCallback)

这段代码有什么问题？

• 回调是异步的，你没法用同步的方式拿到结果
• 生命周期管理要靠自己手动stopScan
• 多个回调组合起来，代码复杂度指数级上升

我在项目中遇到过最离谱的情况：一个同事写了三层嵌套的回调，里面还混着Handler.postDelayed。调试的时候，光看代码就花了半天。

用suspendCancellableCoroutine封装单次回调

先处理最简单的场景：蓝牙开关状态。这个只需要监听一次结果。

suspend fun isBluetoothEnabled(): Boolean = suspendCancellableCoroutine { cont ->
    val adapter = BluetoothAdapter.getDefaultAdapter()
    if (adapter == null) {
        cont.resume(false)
        return@suspendCancellableCoroutine
    }
    cont.resume(adapter.isEnabled)
}

嗯，这个太简单了。我们看个更实际的——连接GATT设备：

suspend fun connectGatt(context: Context, device: BluetoothDevice): BluetoothGatt? {
    return suspendCancellableCoroutine { cont ->
        val gatt = device.connectGatt(context, false, object : BluetoothGattCallback() {
            override fun onConnectionStateChange(gatt: BluetoothGatt?, status: Int, newState: Int) {
                if (newState == BluetoothProfile.STATE_CONNECTED) {
                    cont.resume(gatt)
                } else if (status != GATT_SUCCESS) {
                    cont.resume(null)
                }
            }
        })
        // 如果协程被取消，断开连接
        cont.invokeOnCancellation {
            gatt?.disconnect()
            gatt?.close()
        }
    }
}

这里有个关键点：invokeOnCancellation。我曾经踩过一个坑——用户退出页面时协程被取消了，但蓝牙连接还挂着，导致下次进来连接失败。加上这个清理逻辑，就安全多了。

用callbackFlow实现扫描流

单次回调搞定了，那多次回调呢？比如蓝牙扫描，会持续返回设备信息。

这时候就该callbackFlow上场了。说白了，它能把多次回调包装成一个冷流，每次有新设备就发射数据。

fun scanDevicesFlow(scanFilters: List<ScanFilter>? = null): Flow<ScanResult> = callbackFlow {
    val scanner = BluetoothAdapter.getDefaultAdapter()?.bluetoothLeScanner
    if (scanner == null) {
        close(IllegalStateException("蓝牙不可用"))
        return@callbackFlow
    }

    val scanCallback = object : ScanCallback() {
        override fun onScanResult(callbackType: Int, result: ScanResult?) {
            result?.let { trySend(it) }
        }
        override fun onScanFailed(errorCode: Int) {
            close(ScanException("扫描失败，错误码：$errorCode"))
        }
    }

    val settings = ScanSettings.Builder()
        .setScanMode(ScanSettings.SCAN_MODE_LOW_LATENCY)
        .build()

    scanner.startScan(filters, settings, scanCallback)

    // 协程取消时停止扫描
    awaitClose {
        scanner.stopScan(scanCallback)
    }
}

你看，整个扫描逻辑被封装成了一个Flow。调用方只需要：

viewModelScope.launch {
    scanDevicesFlow()
        .debounce(300)  // 防抖，避免界面刷新太快
        .distinctUntilChanged()  // 去重
        .collect { result ->
            // 更新UI
        }
}

为什么用trySend而不是send？因为callbackFlow的回调可能来自不同线程，send是挂起函数，在非协程上下文调用会出问题。trySend是非挂起的，更适合回调场景。

Flow操作符实战：去重、过滤、超时

有了Flow，蓝牙数据的处理就变得非常灵活了。我常用的几个操作符：

场景	操作符	说明
设备重复出现	distinctUntilChanged	按设备地址去重
信号太弱不要	filter	过滤RSSI小于-80的设备
扫描超时	timeout	超过10秒没结果就抛异常
合并多个设备	scan	累加设备列表

举个例子，实现一个带超时和信号过滤的扫描：

scanDevicesFlow()
    .filter { it.rssi > -80 }
    .distinctUntilChanged { old, new -> old.device.address == new.device.address }
    .timeout(10.seconds)
    .catch { e ->
        emit(null)  // 超时或错误时发射空值
    }
    .collect { result ->
        // 处理结果
    }

我个人习惯把这种组合逻辑封装成扩展函数，方便复用：

fun Flow<ScanResult>.validDeviceScan(): Flow<ScanResult> = this
    .filter { it.rssi > -80 }
    .distinctUntilChanged { old, new -> old.device.address == new.device.address }

Room存储蓝牙设备信息

扫描到的设备信息，总不能每次都重新扫描吧？我建议用Room存起来。这样下次打开App，可以直接从数据库加载历史设备。

先定义实体：

@Entity(tableName = "bluetooth_devices")
data class BluetoothDeviceEntity(
    @PrimaryKey
    val address: String,
    val name: String?,
    val rssi: Int,
    val lastSeenTime: Long,
    val isBonded: Boolean
)

DAO接口：

@Dao
interface BluetoothDeviceDao {
    @Query("SELECT * FROM bluetooth_devices ORDER BY lastSeenTime DESC")
    fun getAllDevices(): Flow<List<BluetoothDeviceEntity>>

    @Insert(onConflict = OnConflictStrategy.REPLACE)
    suspend fun insertDevice(device: BluetoothDeviceEntity)

    @Query("DELETE FROM bluetooth_devices WHERE lastSeenTime < :threshold")
    suspend fun deleteOldDevices(threshold: Long)
}

注意DAO返回的是Flow，这意味着数据库变化时，UI会自动更新。配合扫描流一起用，效果很好：

// 扫描到设备后，同时写入数据库
scanDevicesFlow()
    .collect { result ->
        val entity = BluetoothDeviceEntity(
            address = result.device.address,
            name = result.device.name,
            rssi = result.rssi,
            lastSeenTime = System.currentTimeMillis(),
            isBonded = result.device.bondState == BluetoothDevice.BOND_BONDED
        )
        dao.insertDevice(entity)
    }

这里有个细节：OnConflictStrategy.REPLACE。同一个设备可能被扫描到多次，用地址作为主键，后一次覆盖前一次，保证数据是最新的。

整体架构图

把上面这些串起来，整个蓝牙数据流的架构是这样的：

从这张图能看出来，整个流程分四层：

1. 硬件层：蓝牙芯片和系统服务，产生原始回调
2. 协程封装层：用suspendCancellableCoroutine和callbackFlow把回调转成挂起函数和Flow
3. Flow处理层：用各种操作符过滤、去重、超时处理
4. 存储层：Room持久化，同时用Flow把数据库变化推给UI

每一层各司其职，互不干扰。你想想看，如果哪天蓝牙协议升级了，只需要改最底层的封装，上层的业务逻辑完全不用动。这就是分层设计的好处。

---

公众号:蓝海资料掘金营，微信deep3321