Android 沾包处理，以串口接入为例 (usb-serial-for-android)

1. 前言

我们在通过串口、tcp、UDP等方式接收协议的时候，由于单次接收数据有限，导致一条命令可能被分割成多次进行接收。
这种情况下，就需要进行沾包处理，使多次接收的数据，合并成一条数据。本文通过博主本人一个真实的工作案例，实例讲解Android串口的接入和对于沾包的处理。

2. 协议

我们以下方这个协议为例
这是个串口协议，Android设备通过监听串口，读取到具体的数据

可以看到，前导帧为1个字节，每当读取到0xAA，就代表一条命令的开始。
第二个字节是长度，占1个字节，表示内容部分占用多少个字节。
最后一个字节用特定的算法，将命令的前面部分进行计算后得到的值，用来校验这条命令是否正确。

• 如果命令正确，那就正常处理
• 如果命令错误，就作丢弃处理

3. 验证串口硬件是否正常

可以在平板或手机上下载usb调试宝，设置好波特率 (比如115200，这个根据串口设备设置)，然后即可监听到串口发送的数据了。

4. 串口接入

我们这里使用了usb-serial-for-android这个串口库

4.1 添加Jitpack仓库

repositories {    ...    Maven { url 'https://jitpack.io' }}

4.2 添加usb-serial-for-android依赖

implementation 'com.GitHub.mik3y:usb-serial-for-android:3.4.6'

4.3 获取UsbManager

val usbManager = context.getSystemService(Context.USB_SERVICE) as UsbManager

4.4 判断是否有权限

fun hasPermission(): Boolean {    val driver = getDriver() ?: return false    return usbManager.hasPermission(driver.device)}private fun getDrivers(): MutableList<UsbSerialDriver> {    return UsbSerialProber.getDefaultProber().findAllDrivers(usbManager)}private fun getDriver(): UsbSerialDriver? {    val availableDrivers = getDrivers()    if (availableDrivers.isEmpty()) {        log("availableDrivers is empty.")        return null    }    return availableDrivers[0]}

4.5 请求权限

如果没有权限，需要先申请权限，这一步很主要，要不然后面肯定是读取不到串口的数据的。

fun requestPermission() {    val driver = getDriver() ?: return    val flags =        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) PendingIntent.FLAG_IMMUTABLE else 0    val permissionIntent = PendingIntent.getBroadcast(        context,        0,        Intent("com.android.example.USB_PERMISSION"),        flags    )    usbManager.requestPermission(driver.device, permissionIntent)}

4.6 打开设备

val driver = getDriver() ?: returnval connection = usbManager.openDevice(driver.device) ?: returnlog("connection:$connection")port = driver.ports[0] // Most devices have just one port (port 0)port?.open(connection)port?.setParameters(params.baudRate, params.dataBits, params.stopBits, params.parity)usbIoManager = SerialInputOutputManager(port, this)usbIoManager.start()

注意这里SerialInputOutputManager有个监听，onNewData就是处理接收数据的地方了。

override fun onNewData(data: ByteArray?) {    //当接收到数据}override fun onRunError(e: Exception?) {    //当运行出错}

4.7 关闭设备

当我们要退出App的时候，需要去关闭串口

fun closeDevice() {    port?.close()    port = null}

5. 沾包处理

当我们在onNewData里，我们需要进行沾包处理。
这里我处理沾包的一个思路是在onNewData接收到的数据，存储到一个地方，然后另起一个线程，在那个线程中，再去读取数据。这样，就可以很好地规避在onNewData里，一股脑给到一个ByteArray数组，导致的拆解数据，处理多种异常情况的问题了。

而onNewData接收到的数据，我们可以存储到Queue(队列)，队列的特性是先进先出(通常但并非一定)，这样就可以确保我们先接收到的数据先被读取处理，并且也简化了处理的流程。

5.1 常见的Queue

常见的Queue有这几种，我们这里选用的是LinkedBlockingQueue，没有数据的时候，它具有自动阻塞的能力。

• ArrayBlockingQueue : 数组实现的有界队列，会自动阻塞，根据调用api不同，有不同特性，当队列容量不足时，有阻塞能力
  • boolean add(E e)：在容量不足时，抛出异常。
  • void put(E e)：在容量不足时，阻塞等待。
  • boolean offer(E e)：不阻塞，容量不足时返回false，当前新增数据操作放弃。
  • boolean offer(E e, long timeout, TimeUnit unit）：容量不足时，阻塞times时长（单位为timeunit），如果在阻塞时长内，有容量空闲，新增数据返回true。如果阻塞时长范围内，无容量空闲，放弃新增数据，返回false。
• LinkedBlockingQueue:链式队列，队列容量不足或为0时自动阻塞
  • void put(E e)：自动阻塞，队列容量满后，自动阻塞。
  • E take()：自动阻塞，队列容量为0后，自动阻塞。
• ConcurrentLinkedQueue : 基础链表同步队列
  • boolean offer(E e)：入队。
  • E peek()：查看queue中的首数据。
  • E poll()：取出queue中的首数据。
• DelayQueue: 延时队列，根据比较机制，实现自定义处理顺序的队列。常用于定时任务，如：定时关机。
  • int compareTo(Delayed o)：比较大小，自动升序。
  • 比较方法建议和getDelay方法配合完成。如果在DelayQueue是需要按时完成的计划任务，必须配合getDelay方法完成。
  • long getDelay(TimeUnit unit)：获取计划时长的方法，根据参数TimeUnit来决定，如何返回结果值。
• LinkedTransferQueue : 转移队列
  • boolean add(E e)：队列会保存数据，不做阻塞等待。
  • void transfer(E e)：是TransferQueue的特有方法。必须有消费者（take()方法调用者）。如果没有任意线程消费数据，transfer方法阻塞。一般用于处理及时消息。
• SynchronousQueue : 同步队列，容量为0，是特殊的TransferQueue，必须先有消费线程等待，才能使用的队列。
  • boolean add(E e)：父类方法，无阻塞，若没有消费线程阻塞等待数据，则抛出异常。
  • put(E e)：有阻塞，若没有消费线程阻塞等待数据，则阻塞。

详细关于Queue的介绍，详见 https://blog.csdn.net/qq_37050329/article/details/116295082

5.2 启动线程

在打开串口的时候，我们去启动另一个线程。这里我使用到了线程池，newSingleExecutor是一个单线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序执行。

private val dataQueue = LinkedBlockingQueue<Byte>()private val singleExecutor: Executor by lazy {    Executors.newSingleThreadExecutor()}private val readRunnable = Runnable {    //TODO 具体实现}singleExecutor.execute(readRunnable) //启动线程

5.3 定义Cmd用来接收命令

class Cmd {    compaNIOn object {        const val PREAMBLE: Byte = 0xAA.toByte()    }    var preamble: Byte? = null    var length: Byte = -1    var payload = ArrayList<Byte>()    var checkSum: Byte? = null    fun clear() {        preamble = null        length = -1        payload.clear()        checkSum = null    }}

5.4 进行沾包处理

在readRunnable中，我们去读取dataQueue的数据，当dataQueue没有数据的时候，会进行阻塞，这样就避免了性能的损耗。

val byte = dataQueue.take()

接着，如果我们读到前导帧，就假设读取到了一条命令，按顺序依次读取长度、内容、校验，所有的值都读取到后，需要对校验值checkSum做效验，具体校验的算法根据协议约定来。
命令校验通过后，就可以取到内容，转化为json，进一步做业务逻辑处理了。

val PREAMBLE: Byte = 0xAA.toByte()if (byte == PREAMBLE) { //前导帧    cmd = Cmd()    cmd.preamble = PREAMBLE    log("前导帧:0x${HexUtil.toByteString(PREAMBLE)}")    cmd.length = dataQueue.take()    log("长度:${cmd.length}")    readPayload(dataQueue)    log("内容:${HexUtil.bytesToHexString(cmd.payload.toByteArray())}")    val checkSum = dataQueue.take()    cmd.checkSum = checkSum    log("校验:0x${HexUtil.toByteString(checkSum)}")    //TODO 需要对checkSum进行校验，判断命令是否正确val json = String(cmd.payload.toByteArray()) //内容转换为Json，这里可以做进一步逻辑处理    cmd.clear()} else {    Log.e("Heiko", "被抛弃:0x${HexUtil.toByteString(byte)}")}private fun readPayload(dataStack: LinkedBlockingQueue<Byte>) {    for (i in 0 until cmd.length) {        cmd.payload.add(dataStack.take())    }}

至此，对于沾包的处理就完成了

6. 附录

6.1 封装的串口工具类

附上基于usb-serial-for-android封装好的串口工具类完整代码

class UsbSerialManager(    private val context: Context,    private val params: UsbSerialParams,    private val receiver: (String) -> Unit) :    SerialInputOutputManager.Listener {    private var port: UsbSerialPort? = null    private lateinit var usbIoManager: SerialInputOutputManager    private val dataQueue = LinkedBlockingQueue<Byte>()    private var cmd: Cmd = Cmd()    private val singleExecutor: Executor by lazy {        Executors.newSingleThreadExecutor()    }    private val readRunnable: Runnable    private val usbManager = context.getSystemService(Context.USB_SERVICE) as UsbManager    init {        readRunnable = Runnable {            while (port?.isOpen == true || dataQueue.isNotEmpty()) {                val byte = dataQueue.take()                if (byte == PREAMBLE) { //前导帧                    cmd = Cmd()                    cmd.preamble = PREAMBLE                    log("前导帧:0x${HexUtil.toByteString(PREAMBLE)}")                    cmd.length = dataQueue.take()                    log("长度:${cmd.length}")                    readPayload(dataQueue)                    log("payload:${HexUtil.bytesToHexString(cmd.payload.toByteArray())}")                    val checkSum = dataQueue.take()                    cmd.checkSum = checkSum                    log("校验:0x${HexUtil.toByteString(checkSum)}")                    receiver.invoke(String(cmd.payload.toByteArray()))                    cmd.clear()                } else {                    Log.e("Heiko", "被抛弃:0x${HexUtil.toByteString(byte)}")                }            }        }    }    private fun readPayload(dataStack: LinkedBlockingQueue<Byte>) {        for (i in 0 until cmd.length) {            cmd.payload.add(dataStack.take())        }    }    fun requestPermission() {        val driver = getDriver() ?: return        val flags =            if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) PendingIntent.FLAG_IMMUTABLE else 0        val permissionIntent = PendingIntent.getBroadcast(            context,            0,            Intent("com.android.example.USB_PERMISSION"),            flags        )        usbManager.requestPermission(driver.device, permissionIntent)    }    private fun getDrivers(): MutableList<UsbSerialDriver> {        return UsbSerialProber.getDefaultProber().findAllDrivers(usbManager)    }    private fun getDriver(): UsbSerialDriver? {        val availableDrivers = getDrivers()        if (availableDrivers.isEmpty()) {            log("availableDrivers is empty.")            return null        }        return availableDrivers[0]    }    fun hasPermission(): Boolean {        val driver = getDriver() ?: return false        return usbManager.hasPermission(driver.device)    }    fun openDevice() {        if (port?.isOpen == true) {            log("port is opened.")            return        }        val driver = getDriver() ?: return        debugLogDrivers()        val connection = usbManager.openDevice(driver.device) ?: return        log("connection:$connection")        port = driver.ports[0] // Most devices have just one port (port 0)        port?.open(connection)        port?.setParameters(params.baudRate, params.dataBits, params.stopBits, params.parity)        usbIoManager = SerialInputOutputManager(port, this)        usbIoManager.start()        singleExecutor.execute(readRunnable)        log("usbIoManager.start")    }    private fun debugLogDrivers() {        if (params.debug) {            getDrivers().forEach {                val device = it.device                log(                    "deviceId:${device.deviceId} " +" deviceName:${device.deviceName} " +" deviceProtocol:${device.deviceProtocol} " +" productName:${device.productName}" +" productId:${device.productId}" +" manufacturerName:${device.manufacturerName}" +" configurationCount:${device.configurationCount}" +" serialNumber:${device.serialNumber}" +" vendorId:${device.vendorId}"                )            }        }    }    fun closeDevice() {        port?.close()        port = null    }    private fun receive(data: ByteArray?) {        log("receive:${HexDump.dumpHexString(data)}", "RRRRRRR")        if (data == null) return        for (byte in data) {            dataQueue.put(byte)        }    }    override fun onNewData(data: ByteArray?) {        receive(data)    }    override fun onRunError(e: Exception?) {        log("onRunError:${e?.message}")    }    private fun log(message: String, tag: String = "Heiko") {        Log.i(tag, message)    }}class Cmd {    companion object {        const val PREAMBLE: Byte = 0xAA.toByte()    }    var preamble: Byte? = null    var length: Byte = -1    var payload = ArrayList<Byte>()    var checkSum: Byte? = null    fun clear() {        preamble = null        length = -1        payload.clear()        checkSum = null    }}

6.2 字节数组转字符串工具类

附上字节数组转字符串工具类

public class HexUtil {    public static byte[] hexStringToBytes(String hexString) {        if (hexString == null || hexString.equals("")) {            return null;        }        hexString = hexString.toUpperCase();        int length = hexString.length() / 2;        char[] hexChars = hexString.toCharArray();        byte[] d = new byte[length];        for (int i = 0; i < length; i++) {            int pos = i * 2;            d[i] = (byte) (charToByte(hexChars[pos]) << 4 | charToByte(hexChars[pos + 1]));        }        return d;    }    private static byte charToByte(char c) {        return (byte) "0123456789ABCDEF".indexOf(c);    }    public static String bytesToHexString(byte[] b) {        if (b.length == 0) {            return null;        }        StringBuilder sb = new StringBuilder("");        for (int i = 0; i < b.length; i++) {            int value = b[i] & 0xFF;            String hv = Integer.toHexString(value);            if (hv.length() < 2) {                sb.append(0);            }            sb.append("0x").append(hv).append(" ");        }        return sb.toString();    }    public static String toByteString(byte b) {        String hex = Integer.toHexString(b & 0xFF);        if (hex.length() == 1) {            hex = '0' + hex;        }        return hex.toUpperCase();    }}

来源地址：https://blog.csdn.net/EthanCo/article/details/129194519