一种通用 HEX 协议解析器的简单实现

背景描述

在嵌入式开发中，MCU 通常会与其它模块进行数据交互。简单来说，常见的数据大概有两种：一种是明文字符串流，比如类似 AT 指令；另一种是 HEX 字节流。

本文实现一种能够较为广泛使用的通用协议解析器，用来处理 HEX 数据。

本文按 protocol_lib 仓库当前最新源码整理，对应提交为 5b0c86b。和最初版本相比，现在输入、输出缓冲区可以按实例配置，多协议之间的匹配状态也已经完全分开。

一、前言

HEX 协议基本都是私有定制的，但它们又有一定共性。比如下面这一种：

帧头	帧类型	数据长度	数据	校验	帧尾
A5 5A AA 55	1 byte	1 byte	n byte	XOR	0D 0A 0D 0A

一般 HEX 协议的数据流都类似这种，或者是它的变种。这些数据可以通过 UART、网口、蓝牙、2.4G、SPI 等等等等进行传输。简单点，我们就串口而言，解析方式大致有两种。

一种是利用串口空闲中断，逐帧接收处理。这种方式最简单，CPU 开销也小，但不是每一种 MCU 都有可靠的串口空闲中断。另外，发送过程中还可能出现黏包，比如网口通信时由于网络延迟，几帧数据会一起收到。

另一种就是本文要实现的：把收到的数据先丢进环形缓冲区，再逐字节匹配并解析。它不怕拆包，也不怕多帧一起到达；配合 DMA 或接收缓冲区使用，CPU 开销也还能接受，只是代码写起来稍微麻烦一点。

于是本文应运而生。直接用这个模块，几分钟就能把协议接收部分搭起来~~~

二、实现思路

简单的才是稳定好用的。这个模块要足够容易移植，如果是我用，我最关心的主流程只有下面几件事：

初始化：把帧头、帧尾、长度计算方式、校验方式和缓冲区告诉解析器。
放入数据：UART、SPI 或网口收到多少数据，就往解析器里放多少。
获取完整帧：在主循环或任务中轮询，解析器自己完成同步、组帧、帧尾匹配和校验。
提供心跳：通常每 1 ms 调用一次，让异常帧能够通过空闲超时恢复。

1、提取协议共性

要把一帧数据完整提取出来，首先需要匹配帧头；然后要知道完整帧长，才能判断数据是否收齐并找到帧尾；最后再对整帧进行校验。

所以，当前版本的协议描述结构如下：

typedef uint8_t b_check_cb_t(uint8_t *buffer, uint16_t len);
typedef uint16_t b_get_frame_len_cb_t(uint8_t *buffer, uint16_t len);

typedef struct
{
    const uint8_t *pname;
    const char *head;
    const char *end;
    uint16_t head_len;
    uint16_t end_len;
    b_get_frame_len_cb_t *get_frame_len_cb;
    b_check_cb_t *check_cb;
    uint8_t *in_frame_buffer;
    uint8_t *out_frame_buffer;
    uint16_t in_buffer_len;
    uint16_t out_buffer_len;
    uint8_t log_level;
} b_frame_init_type;

用户只需要描述帧头、帧尾，实现“获取完整帧长”和“校验完整帧”两个回调，再给每个解析器实例准备输入、输出缓冲区即可。

这里有一个很重要的约定：get_frame_len_cb 返回的是完整帧长度，帧头、类型、长度字节、数据、校验和帧尾全部都要算进去；当前数据还不足以判断长度时返回 0。

仍然按上面的协议举例。数据长度在 buffer[5]，那么完整帧长应当这样计算：

static uint16_t get_frame_len_cb(uint8_t *buffer, uint16_t len)
{
    if (len < 6)
    {
        return 0;
    }

    return (uint16_t)(4 + 1 + 1 + buffer[5] + 1 + 4);
}

下面假设 XOR 的计算范围是“帧类型 + 数据长度 + 数据”，校验字节放在帧尾前面：

static uint8_t check_frame_cb(uint8_t *buffer, uint16_t len)
{
    uint8_t xor_value = 0;

    if (len < 11)
    {
        return B_ERROR;
    }

    for (uint16_t i = 4; i < len - 5; i++)
    {
        xor_value ^= buffer[i];
    }

    return (xor_value == buffer[len - 5]) ? B_SUCCESS : B_ERROR;
}

校验成功要返回 B_SUCCESS，失败返回 B_ERROR。当前源码中，即使协议本身没有 CRC 或 Checksum，也不能简单把 check_cb 设成 NULL，因为 b_frame_check_get 只会在校验回调存在并返回成功时交出完整帧。没有校验字段时，写一个最基本的长度或字段合法性检查就好。

2、API 设计

这套解析器最开始只能同时支持一种协议。如果一个项目里出现两种不同的 HEX 协议，难道还要搞两份 .c、两份 .h，然后再改一遍函数名吗？刚毕业工作时我还真这样干过……

所以这里稍微稍微借鉴了一点面向对象的思路：一个 b_frame_type 就是一个独立的协议解析器实例。

typedef struct
{
    b_frame_init_type frame_init;
    ring_buf_t _frame_ring;
    uint32_t _idie_timer;
    uint32_t _systick;
    uint8_t head_match_flg;
} b_frame_type;

协议描述、环形缓冲区、空闲计时和帧头匹配状态全部属于实例。需要同时解析多少种协议，就创建多少个 b_frame_type，每个实例再配一套独立缓冲区。这样多协议之间不会互相影响。

底层环形缓冲区目前主要用到这些接口：

typedef struct
{
    unsigned char *buf;
    unsigned int size;
    unsigned int front;
    unsigned int rear;
} ring_buf_t;

bool ring_buf_init(ring_buf_t *r, unsigned char *buf, unsigned int size);
void ring_buf_clr(ring_buf_t *r);
unsigned int ring_buf_len(ring_buf_t *r);
unsigned int ring_buf_put(ring_buf_t *r, unsigned char *buf, unsigned int len);
unsigned int ring_buf_get(ring_buf_t *r, unsigned char *buf, unsigned int len);
unsigned int ring_buf_check_get(ring_buf_t *r, unsigned char *buf, unsigned int len);
unsigned int ring_buf_clr_len(ring_buf_t *r, unsigned int len);
unsigned char *ring_buf_peek(ring_buf_t *r, unsigned int offset, unsigned int *len);

ring_buf_get 会在读取后移动队列指针，ring_buf_check_get 只偷看数据，不消费；ring_buf_peek 则按偏移查看环形缓冲区中的一个字节。后面匹配帧头时会用到它。

解析器的公共 API 如下：

uint8_t b_frame_init(b_frame_type *pframe,
                     b_frame_init_type *pframeinit);

void b_frame_idie_timer(b_frame_type *pframe);

uint8_t b_frame_put(b_frame_type *pframe,
                    uint8_t *dat,
                    uint32_t len);

const uint8_t *b_frame_check_get(b_frame_type *pframe,
                                 uint16_t *len);

void b_frame_fifo_clear(b_frame_type *pframe);

uint32_t b_frame_fifo_get(b_frame_type *pframe,
                          uint8_t *dat,
                          uint32_t len,
                          uint32_t timeout);

正常的协议解析只需要前四个。b_frame_fifo_clear 用于主动清空接收 FIFO，b_frame_fifo_get 更像一个阻塞式读取指定长度数据的辅助接口。

3、初始化和使用

每个协议实例都要准备独立的输入缓冲区和输出缓冲区。输入缓冲区给 ringbuffer 使用，建议大小直接取 2 的 N 次幂；输出缓冲区必须能够放下最大完整帧。

#include "b_protocol_core.h"

static b_frame_type frame_engine;
static uint8_t frame_in_buffer[256];
static uint8_t frame_out_buffer[256];

static void protocol_init(void)
{
    b_frame_init_type frame_init = {
        .pname = (const uint8_t *)"engine",
        .head = "\xA5\x5A\xAA\x55",
        .end = "\x0D\x0A\x0D\x0A",
        .head_len = 4,
        .end_len = 4,
        .get_frame_len_cb = get_frame_len_cb,
        .check_cb = check_frame_cb,
        .in_frame_buffer = frame_in_buffer,
        .out_frame_buffer = frame_out_buffer,
        .in_buffer_len = sizeof(frame_in_buffer),
        .out_buffer_len = sizeof(frame_out_buffer),
        .log_level = 0,
    };

    (void)b_frame_init(&frame_engine, &frame_init);
}

收到数据后，不用管这次收到的是半帧、一帧还是好几帧，直接调用 b_frame_put：

void protocol_rx(uint8_t *data, uint16_t len)
{
    (void)b_frame_put(&frame_engine, data, len);
}

然后在主循环或任务中轮询：

static void protocol_poll(void)
{
    uint16_t frame_len = 0;
    const uint8_t *frame =
        b_frame_check_get(&frame_engine, &frame_len);

    if (frame == NULL)
    {
        return;
    }

    switch (frame[4])
    {
        /* 根据帧类型处理数据 */
        default:
            break;
    }
}

返回的 frame 指向实例自己的 out_frame_buffer，下一次解析可能覆盖它。如果后面的业务要长期保存这一帧，记得及时拷贝。

三、代码实现

1、日志

协议解析最难受的地方通常不是写，而是出问题后不知道错在哪，所以保留一点日志还是很有必要的。

当前版本支持两层开关：EN_FRAME_DEBUG 用来全局编译关闭；打开后，每个协议实例还可以通过 log_level 控制输出等级。

#define EN_FRAME_DEBUG 0

#define FRAME_LOG(p, lvl, ...)                         \
    do                                                 \
    {                                                  \
        if ((p) && (p)->frame_init.log_level >= (lvl)) \
        {                                              \
            xprintf(__VA_ARGS__);                      \
        }                                              \
    } while (0)

#define FRAME_RAW_INFO_PRINTF(p, ...) FRAME_LOG(p, 1, __VA_ARGS__)
#define FRAME_LOG_INFO_PRINTF(p, ...) FRAME_LOG(p, 2, __VA_ARGS__)

EN_FRAME_DEBUG 为 0 时，这些日志宏会被编译为空；打开时工程需要提供 xprintf。嗯，够简单了~

2、初始化

初始化主要做三件事：检查必要参数，把协议描述复制到实例里，然后用用户提供的输入缓冲区初始化 ringbuffer。

uint8_t b_frame_init(b_frame_type *pframe,
                     b_frame_init_type *pframeinit)
{
    uint8_t err = 0;
    if ((!pframeinit) || (!pframe) ||
        (!pframeinit->in_frame_buffer) ||
        (!pframeinit->out_frame_buffer))
    {
        return B_ERROR;
    }

    if (pframeinit->head == NULL)
        pframeinit->head_len = 0;
    if (pframeinit->end == NULL)
        pframeinit->end_len = 0;

    pframe->frame_init.pname = pframeinit->pname;
    pframe->frame_init.check_cb = pframeinit->check_cb;
    pframe->frame_init.end = pframeinit->end;
    pframe->frame_init.end_len = pframeinit->end_len;
    pframe->frame_init.get_frame_len_cb =
        pframeinit->get_frame_len_cb;
    pframe->frame_init.head = pframeinit->head;
    pframe->frame_init.head_len = pframeinit->head_len;
    pframe->frame_init.in_frame_buffer =
        pframeinit->in_frame_buffer;
    pframe->frame_init.out_frame_buffer =
        pframeinit->out_frame_buffer;
    pframe->frame_init.in_buffer_len =
        pframeinit->in_buffer_len;
    pframe->frame_init.out_buffer_len =
        pframeinit->out_buffer_len;
    pframe->frame_init.log_level = pframeinit->log_level;
    pframe->head_match_flg = 0;

    err = ring_buf_init(&pframe->_frame_ring,
                        pframe->frame_init.in_frame_buffer,
                        pframe->frame_init.in_buffer_len);
    b_frame_fifo_clear(pframe);

    return err ? B_SUCCESS : B_ERROR;
}

frame_init 结构体本身可以是局部变量，因为配置已经逐字段复制进 frame_engine。不过这些字段里的指针只是被保存，并没有深拷贝；pname、head、end 以及输入、输出缓冲区指向的数据都必须一直有效。示例里的名称、帧头和帧尾是字符串常量，两个缓冲区则是静态数组，正好满足这个条件。

如果协议没有帧头，就让 head = NULL；没有帧尾，就让 end = NULL。初始化函数会自动把对应长度置 0。

3、put 函数

这个函数基本没什么好讲的，就是把数据丢进队列，再加一点安全处理。

uint8_t b_frame_put(b_frame_type *pframe,
                    uint8_t *dat,
                    uint32_t len)
{
    if (!dat || len == 0)
    {
        return B_ERROR;
    }

    pframe->_idie_timer = 0;

    uint32_t putlen =
        ring_buf_put(&pframe->_frame_ring, dat, len);

    if (putlen != len)
    {
        ring_buf_clr(&pframe->_frame_ring);
        return B_ERROR;
    }

    return B_SUCCESS;
}

当前实现要求本次数据全部写入。空间不够而只能部分写入时，会直接清空输入 FIFO 并返回错误，避免留下半截、不知道从哪里开始的数据。

4、匹配帧头

最新版本不再一边比较一边直接拿走所有候选字节，而是通过 ring_buf_peek 查看；完整帧头匹配成功后才一次性消费帧头。遇到不匹配就丢掉一个字节，再从新的队首重新开始。

static uint8_t b_check_head(b_frame_type *pframe)
{
    if (pframe->frame_init.head_len == 0)
        return B_SUCCESS;

    uint8_t i = 0;
    uint8_t *pd = NULL;
    uint16_t len = ring_buf_len(&pframe->_frame_ring);

    if (len < (pframe->frame_init.head_len +
               pframe->frame_init.end_len + 1))
    {
        return B_ERROR;
    }

    do
    {
        pd = ring_buf_peek(&pframe->_frame_ring, i, NULL);
        if (pd == NULL)
            break;

        if (*pd == (uint8_t)pframe->frame_init.head[i])
        {
            if (++i == pframe->frame_init.head_len)
            {
                ring_buf_clr_len(&pframe->_frame_ring,
                                 pframe->frame_init.head_len);
                return B_SUCCESS;
            }
        }
        else
        {
            ring_buf_clr_len(&pframe->_frame_ring, 1);
            i = 0;
        }
    } while (1);

    return B_ERROR;
}

这种写法还有一个好处：遇到 AA AA 55 这种和帧头局部重叠的数据时，每次只丢一个字节，不会轻易跨过真正的帧头。

5、解析完整帧

b_frame_check_get 是整个模块的核心，流程大致如下：

当前实例还没有匹配到帧头时，调用 b_check_head。
帧头匹配成功后，把帧头复制到 out_frame_buffer，并置位实例自己的 head_match_flg。
用 ring_buf_check_get 偷看剩余数据，不移动输入 FIFO。
调用 get_frame_len_cb，得到包含帧头和帧尾的完整帧长。
数据不够就继续等；长时间没有新数据则复位匹配状态。
数据够了以后检查帧尾，再调用 check_cb 校验完整帧。
校验通过才真正消费 FIFO 中剩余帧数据，并返回 out_frame_buffer。

其中最关键的长度处理是下面这段：

uint16_t data_len =
    ring_buf_check_get(
        &pframe->_frame_ring,
        &pframe->frame_init.out_frame_buffer[
            pframe->frame_init.head_len],
        pframe->frame_init.out_buffer_len -
            pframe->frame_init.head_len);

unhead_frame_len =
    pframe->frame_init.get_frame_len_cb(
        pframe->frame_init.out_frame_buffer,
        data_len + pframe->frame_init.head_len);

if (unhead_frame_len >= pframe->frame_init.head_len)
{
    unhead_frame_len -= pframe->frame_init.head_len;
}

if ((data_len < unhead_frame_len) ||
    (unhead_frame_len == 0))
{
    return NULL;
}

回调返回完整帧长，解析器再减去已经消费掉的帧头长度，得到 FIFO 中还需要多少数据。帧尾匹配和校验通过后，才把这部分数据真正丢出 FIFO：

*len = pframe->frame_init.head_len + unhead_frame_len;

if (pframe->frame_init.check_cb != NULL)
{
    uint8_t err =
        pframe->frame_init.check_cb(
            pframe->frame_init.out_frame_buffer,
            *len);

    if (err == B_SUCCESS)
    {
        pframe->head_match_flg = 0;
        ring_buf_clr_len(&pframe->_frame_ring,
                         unhead_frame_len);
        return pframe->frame_init.out_frame_buffer;
    }
}

pframe->head_match_flg = 0;
return NULL;

head_match_flg 现在是 b_frame_type 的成员，不再是函数里的静态变量。这一点对多协议支持非常关键，否则两个解析器实例轮询时会共用同一个匹配状态，迟早串台。

四、杂项

还记得协议实例里的 _idie_timer 和 _systick 吗？它们主要用来做异常恢复。

我曾经遇到过一种协议，长度字段有 2 个字节，数据量又非常大。某次帧头在数据区里发生冲突，长度字段刚好被解释成 0xFFFF，解析器就会一直等一个永远收不完整的超长帧。哦豁，卡死。

所以模块里模拟了串口空闲超时。每收到一批新数据，_idie_timer 会清零；如果已经算出帧长但后续数据迟迟不来，超过阈值后就放弃本次匹配，重新寻找帧头。

需要给模块提供一个心跳，按 IDIE_TIMER_US 周期调用。默认值是 1000 us，也就是通常每 1 ms 调用一次：

void b_frame_idie_timer(b_frame_type *pframe)
{
    pframe->_idie_timer++;
    pframe->_systick++;
}

实际移植时，还有几个细节值得再强调一遍：

每一种协议使用独立的 b_frame_type、输入缓冲区和输出缓冲区。
in_buffer_len 建议直接使用 64、128、256、512 这类 2 的 N 次幂。
out_buffer_len 必须能放下最大完整帧，并且不能小于 head_len。
get_frame_len_cb 返回完整帧长；信息不足时返回 0。
check_cb 不能为 NULL，成功时返回 B_SUCCESS。
b_frame_check_get 返回的是内部输出缓冲区，长期使用前要及时拷贝。
b_frame_put 空间不足时会清空输入 FIFO，调用方要检查返回值。
打开 EN_FRAME_DEBUG 后需要由工程提供 xprintf。

结尾

okkkkk，基本讲得差不多了。

完整代码放在 Gitee 仓库，本文整理时对应最新提交 5b0c86b。

欢迎点赞关注。如果你在使用这个模块时发现什么 bug，请务必反馈，不胜感激~~~