一、Nanopb简介与优势

1.1 什么是Nanopb

Nanopb是Protocol Buffers的C语言实现，专门为资源受限的嵌入式系统设计：

• 轻量级：代码占用量小（~20KB ROM，~1KB RAM）

• 无动态内存分配：适合无操作系统的嵌入式环境

• 高效编码：二进制编码，数据量小，解析速度快

1.2 为什么选择Nanopb

// 与传统JSON对比 JSON: {"temp":25.5,"humidity":60} // 约30字节 Protobuf: 0x0D 0x00 0x00 0xCC 0x41 0x15 0x00 0x00 0x70 0x42 // 仅10字节

二、环境搭建与配置

2.1 安装Nanopb

# 方法1：下载源码 git clone https://github.com/nanopb/nanopb.git # 方法2：包管理器 # platformio.ini lib_deps = nanopb/Nanopb@^0.4.7

2.2 工程配置示例（基于STM32+FreeRTOS）

makefile

# Makefile配置
CFLAGS += -I$(NANOPB_DIR)
CFLAGS += -DPB_FIELD_16BIT  # 根据消息大小选择

# 链接文件
OBJS += nanopb/pb_encode.o \
        nanopb/pb_decode.o \
        nanopb/pb_common.o \
        protobuf/sensor.pb.o

三、协议设计与代码生成

3.1 定义.proto文件

protobuf

// sensor_data.proto
syntax = "proto2";

package iot;

message SensorData {
    required uint32 timestamp = 1;      // 时间戳
    required float temperature = 2;     // 温度
    required float humidity = 3;        // 湿度
    optional float pressure = 4;        // 气压（可选）
    repeated uint32 adc_values = 5 [max_count = 8];  // ADC值数组
    enum Status {
        OK = 0;
        ERROR = 1;
        CALIBRATING = 2;
    }
    required Status status = 6;
}

message DeviceInfo {
    required bytes device_id = 1 [(nanopb).max_size = 16];
    required string firmware_version = 2 [(nanopb).max_size = 32];
    required uint32 battery_level = 3;
}

message TelemetryPacket {
    required DeviceInfo device = 1;
    repeated SensorData sensors = 2 [max_count = 10];
    required uint32 packet_id = 3;
}

3.2 生成C代码

bash

# 生成.pb.c和.pb.h文件
python nanopb/generator/nanopb_generator.py \
    -I protobuf \
    -D src/generated \
    sensor_data.proto

四、实战应用代码

4.1 数据编码（发送端）

// telemetry_sender.c
#include "sensor_data.pb.h"
#include "pb_encode.h"
#include "uart.h"

// 静态分配缓冲区（避免动态内存）
static uint8_t tx_buffer[512];

bool encode_sensor_data(const SensorData* sensor, uint8_t* buffer, size_t* length) {
    pb_ostream_t stream = pb_ostream_from_buffer(buffer, sizeof(tx_buffer));
    
    if (!pb_encode(&stream, SensorData_fields, sensor)) {
        LOG_ERROR("Encoding failed: %s", PB_GET_ERROR(&stream));
        return false;
    }
    
    *length = stream.bytes_written;
    return true;
}

bool send_telemetry_packet(void) {
    TelemetryPacket packet = TelemetryPacket_init_zero;
    SensorData sensors[2];
    
    // 填充设备信息
    strncpy(packet.device.device_id, "DEV_001", sizeof(packet.device.device_id));
    strncpy(packet.device.firmware_version, "1.2.3", sizeof(packet.device.firmware_version));
    packet.device.battery_level = 85;
    
    // 传感器1数据
    sensors[0].timestamp = HAL_GetTick();
    sensors[0].temperature = 25.5f;
    sensors[0].humidity = 60.0f;
    sensors[0].pressure = 1013.25f;
    sensors[0].status = Status_OK;
    
    // 传感器2数据
    sensors[1].timestamp = HAL_GetTick();
    sensors[1].temperature = 26.1f;
    sensors[1].humidity = 58.5f;
    sensors[1].status = Status_OK;
    
    // 使用回调函数处理数组
    packet.sensors.funcs.encode = &encode_sensor_array;
    packet.sensors.arg = sensors;
    packet.sensors_count = 2;
    packet.packet_id = packet_counter++;
    
    // 编码
    size_t length;
    if (!encode_telemetry_packet(&packet, tx_buffer, &length)) {
        return false;
    }
    
    // 通过UART发送
    return uart_send(tx_buffer, length);
}

// 数组编码回调函数
bool encode_sensor_array(pb_ostream_t *stream, const pb_field_t *field, void * const *arg) {
    SensorData *sensors = (SensorData*)*arg;
    
    for (int i = 0; i < 2; i++) {
        if (!pb_encode_tag_for_field(stream, field))
            return false;
        
        if (!pb_encode_submessage(stream, SensorData_fields, &sensors[i]))
            return false;
    }
    
    return true;
}

4.2 数据解码（接收端）

// command_receiver.c
#include "sensor_data.pb.h"
#include "pb_decode.h"

typedef struct {
    uint8_t buffer[256];
    size_t length;
} CommandBuffer;

bool decode_device_command(const uint8_t* data, size_t length, DeviceCommand* cmd) {
    pb_istream_t stream = pb_istream_from_buffer(data, length);
    
    // 解码消息
    if (!pb_decode(&stream, DeviceCommand_fields, cmd)) {
        LOG_ERROR("Decoding failed: %s", PB_GET_ERROR(&stream));
        return false;
    }
    
    return true;
}

void process_incoming_data(CommandBuffer* buf) {
    DeviceCommand command = DeviceCommand_init_zero;
    
    if (decode_device_command(buf->buffer, buf->length, &command)) {
        switch (command.type) {
            case CommandType_SET_INTERVAL:
                set_sampling_interval(command.interval);
                break;
                
            case CommandType_CALIBRATE:
                start_calibration();
                break;
                
            case CommandType_UPDATE_CONFIG:
                update_device_config(&command.config);
                break;
        }
        
        // 发送确认
        send_command_ack(command.command_id);
    }
}

五、内存优化技巧

5.1 静态内存池管理

c

// memory_pool.h
#define MAX_SENSOR_PACKETS 10
#define PACKET_SIZE 256

typedef struct {
    uint8_t pool[MAX_SENSOR_PACKETS][PACKET_SIZE];
    uint8_t used[MAX_SENSOR_PACKETS];
} PacketMemoryPool;

bool allocate_packet_buffer(uint8_t** buffer) {
    for (int i = 0; i < MAX_SENSOR_PACKETS; i++) {
        if (!used[i]) {
            used[i] = 1;
            *buffer = pool[i];
            return true;
        }
    }
    return false;
}

5.2 Nanopb配置选项

// nanopb配置头文件 nanopb_config.h
#define PB_ENABLE_MALLOC 0      // 禁用动态内存
#define PB_MAX_REQUIRED_FIELDS 32
#define PB_NO_ERRMSG 1          // 减少错误字符串占用

// 根据消息大小优化
#if MESSAGE_SIZE < 256
    #define PB_FIELD_32BIT 0
    #define PB_FIELD_16BIT 1
#else
    #define PB_FIELD_32BIT 1
    #define PB_FIELD_16BIT 0
#endif

六、通信层封装

6.1 MQTT集成示例

// mqtt_nanopb_adapter.c
#include "mqtt_client.h"
#include "sensor_data.pb.h"

typedef struct {
    mqtt_client_t* client;
    const char* topic;
} MQTTContext;

bool publish_sensor_data_mqtt(MQTTContext* ctx, const SensorData* data) {
    uint8_t buffer[128];
    size_t length;
    
    // 编码为protobuf
    if (!encode_sensor_data(data, buffer, &length)) {
        return false;
    }
    
    // 通过MQTT发布
    return mqtt_publish(ctx->client, ctx->topic, buffer, length, MQTT_QOS_1);
}

void mqtt_message_callback(void* arg, const char* topic, 
                          const uint8_t* data, uint32_t len) {
    DeviceCommand command;
    
    if (decode_device_command(data, len, &command)) {
        // 处理命令
        handle_device_command(&command);
    }
}

七、调试与测试

7.1 调试输出

// debug_utils.c
void print_hex_dump(const uint8_t* data, size_t length) {
    printf("Protobuf raw data (%zu bytes):\n", length);
    for (size_t i = 0; i < length; i++) {
        printf("%02X ", data[i]);
        if ((i + 1) % 16 == 0) printf("\n");
    }
    printf("\n");
}

void print_sensor_data(const SensorData* data) {
    printf("SensorData {\n");
    printf("  timestamp: %u\n", data->timestamp);
    printf("  temperature: %.2f\n", data->temperature);
    printf("  humidity: %.2f\n", data->humidity);
    if (data->has_pressure) {
        printf("  pressure: %.2f\n", data->pressure);
    }
    printf("  status: %d\n", data->status);
    printf("}\n");
}

7.2 单元测试

python

# test_nanopb.py
import sensor_data_pb2
import struct

def test_sensor_data_encoding():
    # Python端测试（用于验证嵌入式端编码）
    sensor = sensor_data_pb2.SensorData()
    sensor.timestamp = 123456789
    sensor.temperature = 25.5
    sensor.humidity = 60.0
    sensor.status = sensor_data_pb2.OK
    
    data = sensor.SerializeToString()
    print(f"Encoded size: {len(data)} bytes")
    
    # 解码验证
    sensor2 = sensor_data_pb2.SensorData()
    sensor2.ParseFromString(data)
    assert sensor2.temperature == 25.5

八、最佳实践总结

8.1 设计原则

1. 预分配内存：避免运行时内存分配

2. 固定大小数组：使用max_count和max_size选项

3. 版本兼容：使用optional字段保证向前/向后兼容

4. 错误处理：始终检查pb_encode/pb_decode返回值

8.2 性能优化

// 批量编码优化
bool encode_sensor_batch(SensorData* sensors, int count, 
                        uint8_t* buffer, size_t* total_length) {
    pb_ostream_t stream = pb_ostream_from_buffer(buffer, BUFFER_SIZE);
    *total_length = 0;
    
    for (int i = 0; i < count; i++) {
        size_t msg_len;
        uint8_t msg_buffer[64];
        
        // 编码单个消息
        pb_ostream_t msg_stream = pb_ostream_from_buffer(msg_buffer, sizeof(msg_buffer));
        if (!pb_encode(&msg_stream, SensorData_fields, &sensors[i]))
            return false;
            
        // 添加长度前缀（用于流式传输）
        encode_varint(stream, msg_stream.bytes_written);
        pb_write(&stream, msg_buffer, msg_stream.bytes_written);
        
        *total_length += msg_stream.bytes_written + 
                        varint_size(msg_stream.bytes_written);
    }
    
    return true;
}