1. 前言
我们在通过串口、TCP、UDP等方式接收协议的时候，由于单次接收数据有限，导致一条命令可能被分割成多次进行接收。
这种情况下，就需要进行沾包处理，使多次接收的数据，合并成一条数据。本文通过博主本人一个真实的工作案例，实例讲解Android串口的接入和对于沾包的处理。
2. 协议
我们以下方这个协议为例
这是个串口协议，Android设备通过监听串口，读取到具体的数据
| 前导帧 | 长度 | 内容 | 校验 | |
|---|---|---|---|---|
| 长度 | 1Bit | 1Bit | 0~255Bit | 1Bit |
| 值 | 0xAA | 0~255 | Json | 校验结果 |
可以看到，前导帧为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 UsbManager4.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
