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

不念不忘少年蓝@ 2023-10-09 15:43 51阅读 0赞

### 1. 前言 ###

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

### 2. 协议 ###

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

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

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

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

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

![在这里插入图片描述][d0fb05f3f43342e8aefea278f606a04d.png]

### 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() ?: return
    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()

注意这里`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][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();
        }
    }

#### 6.3 UsbSerialParams ####

`UsbSerialParams`是一个参数配置类，附上代码

data class UsbSerialParams(
    	var baudRate: Int, //比如 115200
    	var dataBits: Int, //比如 8
    	var stopBits: Int, //比如 UsbSerialPort.STOPBITS_1
    	var parity: Int, //比如 UsbSerialPort.PARITY_NONE
    	var debug: Boolean
    )

[usb]: https://www.liqucn.com/rj/190984.shtml
[d0fb05f3f43342e8aefea278f606a04d.png]: https://img-blog.csdnimg.cn/d0fb05f3f43342e8aefea278f606a04d.png
[usb-serial-for-android]: https://github.com/mik3y/usb-serial-for-android
[https_blog.csdn.net_qq_37050329_article_details_116295082]: https://blog.csdn.net/qq_37050329/article/details/116295082