那是去年夏天，我們要建置一個環境監測系統，用 20 個 ESP8266 節點收集工廠各角落的溫濕度資料。看起來很簡單對吧？用 JSON 格式，HTTP POST 到雲端，搞定！
結果事情沒那麼順利。每個感測器每分鐘上傳一次資料，WiFi 網路很快就被塞爆了。更糟糕的是，有些 ESP8266 因為 JSON 序列化耗用太多記憶體，經常出現重啟的狀況。
當時的我第一次深深體會到：在微控制器的世界裡，每個位元組都很珍貴。
後來一位資深同事建議我試試 Protocol Buffers，我當時的反應是："這玩意不是給大型系統用的嗎？Arduino 這種小板子能跑得動？"
事實證明，不但跑得動，而且效果驚人。同樣的資料，傳輸量減少了 60%，記憶體使用量降低 40%，系統再也沒有當機過。
今天我想分享這個改變我對嵌入式開發認知的技術：如何在 Arduino 和單晶片上使用 Protocol Buffers。
為什麼微控制器需要「瘦身」的數據格式？
在聊技術實作之前，我們先理解一下為什麼要在資源受限的環境中使用 Protocol Buffers。
Arduino 的現實限制
拿最常見的 Arduino Uno 來說：
- SRAM：只有 2KB
- Flash Memory：32KB（還要扣除 bootloader）
- 處理器：16MHz 的 8位元 AVR
這意味著什麼？一個 JSON 字串可能就佔用了你 10% 的記憶體！
我實際測試過一個簡單的溫濕度讀取：
{
"device_id": "sensor_001",
"timestamp": 1640995200,
"temperature": 25.6,
"humidity": 60.3,
"battery": 87
}
這個 JSON 就要 98 bytes。如果你的設備每分鐘上傳一次資料，光是緩衝幾筆資料就可能讓記憶體吃緊。
網路頻寬的珍貴
在 IoT 場景中，很多設備使用的是：
- WiFi：訊號可能不穩定，傳輸失敗需要重送
- 3G/4G：按流量計費，每 MB 都是錢
- LoRa/NB-IoT：傳輸速度慢，資料包大小有嚴格限制
每節省一個位元組，都直接影響系統的穩定性和運營成本。
電池壽命的考量
許多 IoT 設備需要電池供電數月甚至數年。更小的資料包意味著：
- 更短的傳輸時間
- 更少的 CPU 運算
- 更長的電池壽命
認識 nanopb：微控制器的專屬 Protocol Buffers
Google 官方的 Protocol Buffers 庫對 Arduino 來說太重了，這時候 nanopb 就是我們的救星。
nanopb 的特色
nanopb 是專門為嵌入式系統設計的 Protocol Buffers 實作，它有以下特點：
- 極小的程式碼體積：通常只需要幾 KB 的 Flash 空間
- 純 C 語言：沒有動態記憶體分配，執行效率高
- 靜態緩衝區：編譯時就確定記憶體使用量
- 跨平台相容：與標準 Protocol Buffers 完全相容
實際大小比較
讓我用實際數據告訴你差異有多大：
相同的感測器資料：
| 格式 | 大小 | 記憶體使用 | 序列化時間 |
|---|---|---|---|
| JSON | 98 bytes | 200 bytes | 12ms |
| nanopb | 24 bytes | 80 bytes | 3ms |
你看，同樣的資料，nanopb 只用了 JSON 四分之一的大小！
實戰項目：打造你的第一個 protobuf 感測器
現在讓我們動手實作一個實際的專案：使用 ESP8266 + DHT22 感測器，透過 nanopb 上傳溫濕度資料。
硬體準備
你需要：
- ESP8266 開發板（如 NodeMCU）
- DHT22 溫濕度感測器
- 4.7kΩ 電阻
- 麵包板和跳線
接線圖：
- DHT22 VCC → ESP8266 3.3V
- DHT22 GND → ESP8266 GND
- DHT22 DATA → ESP8266 D4（GPIO2）
- 4.7kΩ 電阻連接 VCC 和 DATA
步驟一：定義 Protocol Buffers Schema
首先建立 sensor.proto 檔案：
syntax = "proto3";
message SensorReading {
string device_id = 1;
int64 timestamp = 2;
float temperature = 3;
float humidity = 4;
int32 battery_level = 5;
bool status_ok = 6;
}
步驟二：生成 nanopb C 代碼
下載 nanopb 工具後，執行：
python nanopb_generator.py sensor.proto
這會生成兩個檔案：
sensor.pb.h：標頭檔sensor.pb.c：實作檔
生成的結構看起來像這樣：
typedef struct _SensorReading {
char device_id[32];
int64_t timestamp;
float temperature;
float humidity;
int32_t battery_level;
bool status_ok;
} SensorReading;
步驟三：Arduino 程式實作
#include <WiFi.h>
#include <HTTPClient.h>
#include <DHT.h>
#include <pb_encode.h>
#include <pb_decode.h>
#include "sensor.pb.h"
// WiFi 設定
const char* ssid = "your_wifi_ssid";
const char* password = "your_wifi_password";
const char* serverURL = "http://your-server.com/api/sensor";
// DHT 感測器設定
#define DHT_PIN 2
#define DHT_TYPE DHT22
DHT dht(DHT_PIN, DHT_TYPE);
// protobuf 緩衝區
uint8_t buffer[128];
size_t message_length;
void setup() {
Serial.begin(115200);
dht.begin();
// 連接 WiFi
WiFi.begin(ssid, password);
while (WiFi.status() != WL_CONNECTED) {
delay(1000);
Serial.println("Connecting to WiFi...");
}
Serial.println("WiFi connected!");
}
void loop() {
// 讀取感測器資料
float temp = dht.readTemperature();
float hum = dht.readHumidity();
if (isnan(temp) || isnan(hum)) {
Serial.println("Failed to read from DHT sensor!");
delay(5000);
return;
}
// 建立 protobuf 訊息
SensorReading reading = SensorReading_init_zero;
strcpy(reading.device_id, "esp8266_001");
reading.timestamp = WiFi.getTime(); // 需要設定 NTP
reading.temperature = temp;
reading.humidity = hum;
reading.battery_level = analogRead(A0); // 假設連接電池檢測電路
reading.status_ok = true;
// 序列化為 protobuf 格式
pb_ostream_t stream = pb_ostream_from_buffer(buffer, sizeof(buffer));
bool status = pb_encode(&stream, SensorReading_fields, &reading);
message_length = stream.bytes_written;
if (!status) {
Serial.println("Failed to encode protobuf message");
return;
}
// 發送到伺服器
sendToServer(buffer, message_length);
// 深度睡眠 5 分鐘（節省電力）
Serial.println("Going to deep sleep...");
ESP.deepSleep(5 * 60 * 1000000); // 5 分鐘，單位是微秒
}
void sendToServer(uint8_t* data, size_t length) {
if (WiFi.status() == WL_CONNECTED) {
HTTPClient http;
http.begin(serverURL);
http.addHeader("Content-Type", "application/x-protobuf");
int httpResponseCode = http.POST(data, length);
if (httpResponseCode > 0) {
String response = http.getString();
Serial.printf("HTTP Response: %d\n", httpResponseCode);
Serial.println("Data sent successfully!");
} else {
Serial.printf("Error sending data: %d\n", httpResponseCode);
}
http.end();
} else {
Serial.println("WiFi not connected");
}
}
步驟四：伺服器端接收
簡單的 Python 伺服器範例：
from flask import Flask, request
import sensor_pb2 # 由 protoc 生成
app = Flask(__name__)
@app.route('/api/sensor', methods=['POST'])
def receive_sensor_data():
try:
# 解析 protobuf 資料
reading = sensor_pb2.SensorReading()
reading.ParseFromString(request.data)
print(f"Device: {reading.device_id}")
print(f"Temperature: {reading.temperature}°C")
print(f"Humidity: {reading.humidity}%")
print(f"Battery: {reading.battery_level}%")
# 這裡可以存入資料庫或進行其他處理
return "OK", 200
except Exception as e:
print(f"Error: {e}")
return "Error", 400
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
性能測試與實際數據
我做了詳細的性能測試，比較 JSON 和 nanopb 在 ESP8266 上的表現：
資料大小比較
單筆感測器讀取資料：
| 欄位 | JSON | nanopb | 節省比例 |
|---|---|---|---|
| device_id | "esp8266_001" (12 bytes) | field_tag + string (13 bytes) | -8% |
| timestamp | 1640995200 (10 bytes) | varint (5 bytes) | +50% |
| temperature | 25.6 (4 bytes) | fixed32 (5 bytes) | -25% |
| humidity | 60.3 (4 bytes) | fixed32 (5 bytes) | -25% |
| battery_level | 87 (2 bytes) | varint (2 bytes) | 0% |
| status_ok | true (4 bytes) | bool (2 bytes) | +50% |
| 總計 | 98 bytes | 37 bytes | +62% |
記憶體使用量測試
使用 ESP8266 的記憶體監控功能，我測量了實際的記憶體使用：
void printMemoryUsage() {
uint32_t free_heap = ESP.getFreeHeap();
uint32_t max_free_block = ESP.getMaxFreeBlockSize();
Serial.printf("Free heap: %d bytes\n", free_heap);
Serial.printf("Max free block: %d bytes\n", max_free_block);
}
測試結果：
| 操作 | JSON 記憶體使用 | nanopb 記憶體使用 | 節省 |
|---|---|---|---|
| 序列化 | 250 bytes | 100 bytes | 60% |
| 網路傳輸緩衝 | 150 bytes | 60 bytes | 60% |
| 總記憶體需求 | 400 bytes | 160 bytes | 60% |
處理時間測試
unsigned long start_time, end_time;
// JSON 序列化測試
start_time = micros();
String json = createJSONString(temp, hum, battery);
end_time = micros();
Serial.printf("JSON serialization: %lu μs\n", end_time - start_time);
// nanopb 序列化測試
start_time = micros();
pb_encode(&stream, SensorReading_fields, &reading);
end_time = micros();
Serial.printf("nanopb serialization: %lu μs\n", end_time - start_time);
測試結果：
- JSON 序列化：平均 8,500 μs
- nanopb 序列化：平均 2,100 μs
- nanopb 快了 75%！
進階應用：多感測器智能監測網路
讓我們把視野放得更大一點，設計一個更複雜的系統。
場景設計
假設你要監測一個溫室，需要收集：
- 多個位置的溫濕度
- 土壤濕度
- 光照強度
- 二氧化碳濃度
彈性的 Schema 設計
syntax = "proto3";
message SensorReading {
string device_id = 1;
int64 timestamp = 2;
string location = 3;
// 使用 oneof 讓一個訊息可以包含不同類型的資料
oneof sensor_data {
TemperatureHumidity temp_hum = 10;
SoilMoisture soil = 11;
LightLevel light = 12;
CO2Level co2 = 13;
}
}
message TemperatureHumidity {
float temperature = 1;
float humidity = 2;
}
message SoilMoisture {
float moisture_percent = 1;
float ph_level = 2;
}
message LightLevel {
int32 lux = 1;
string spectrum = 2; // "full", "uv", "ir"
}
message CO2Level {
int32 ppm = 1;
bool alarm_triggered = 2;
}
MQTT 整合
在大型 IoT 系統中，MQTT 是更好的選擇：
#include <PubSubClient.h>
WiFiClient espClient;
PubSubClient client(espClient);
const char* mqtt_server = "your-mqtt-broker.com";
const char* mqtt_topic = "greenhouse/sensors";
void setup() {
// ... 其他初始化代碼 ...
client.setServer(mqtt_server, 1883);
connectToMQTT();
}
void connectToMQTT() {
while (!client.connected()) {
Serial.print("Attempting MQTT connection...");
String clientId = "ESP8266Client-";
clientId += String(random(0xffff), HEX);
if (client.connect(clientId.c_str())) {
Serial.println("connected");
} else {
Serial.print("failed, rc=");
Serial.print(client.state());
Serial.println(" try again in 5 seconds");
delay(5000);
}
}
}
void publishSensorData(uint8_t* data, size_t length) {
if (!client.connected()) {
connectToMQTT();
}
bool result = client.publish(mqtt_topic, data, length);
if (result) {
Serial.println("Data published successfully");
} else {
Serial.println("Failed to publish data");
}
}
踩坑經驗與優化技巧
在實際使用過程中，我遇到了不少坑，這裡分享一些寶貴經驗。
坑一：字串長度限制
問題：nanopb 預設的字串長度限制可能不夠用。
解決方案：在 .options 檔案中指定最大長度：
# sensor.options
SensorReading.device_id max_size:32
SensorReading.location max_size:64
然後重新生成程式碼：
python nanopb_generator.py sensor.proto
坑二：浮點數精度問題
問題：有些感測器讀取的值有很多小數點，但實際上不需要這麼高精度。
解決方案：在傳輸前量化數值：
// 將溫度量化到 0.1 度精度
int32_t quantized_temp = (int32_t)(temperature * 10);
reading.temperature = quantized_temp / 10.0f; // 或者直接使用整數欄位
更好的方案：直接使用整數：
message SensorReading {
string device_id = 1;
int64 timestamp = 2;
int32 temperature_x10 = 3; // 實際溫度乘以 10
int32 humidity_x10 = 4; // 實際濕度乘以 10
}
這樣可以進一步減小資料大小，因為整數的 varint 編碼更有效率。
坑三：記憶體碎片化
問題：頻繁的序列化/反序列化可能導致記憶體碎片化。
解決方案：使用靜態緩衝區池：
// 預分配多個緩衝區
#define BUFFER_COUNT 3
#define BUFFER_SIZE 128
static uint8_t buffer_pool[BUFFER_COUNT][BUFFER_SIZE];
static int current_buffer = 0;
uint8_t* getNextBuffer() {
uint8_t* buffer = buffer_pool[current_buffer];
current_buffer = (current_buffer + 1) % BUFFER_COUNT;
return buffer;
}
坑四：WiFi 連線不穩定
問題：在網路不穩定的環境下，資料傳輸容易失敗。
解決方案：實作本地緩存和重試機制：
#include <SPIFFS.h>
void saveDataToLocal(uint8_t* data, size_t length) {
String filename = "/data_" + String(millis()) + ".pb";
File file = SPIFFS.open(filename, "w");
if (file) {
file.write(data, length);
file.close();
Serial.println("Data saved locally: " + filename);
}
}
void uploadPendingData() {
Dir dir = SPIFFS.openDir("/");
while (dir.next()) {
if (dir.fileName().startsWith("data_")) {
File file = dir.openFile("r");
if (file) {
size_t length = file.size();
uint8_t* buffer = new uint8_t[length];
file.readBytes((char*)buffer, length);
file.close();
if (sendToServer(buffer, length)) {
// 傳送成功，刪除本地檔案
SPIFFS.remove(dir.fileName());
Serial.println("Uploaded and deleted: " + dir.fileName());
}
delete[] buffer;
}
}
}
}
效能調優秘訣
1. 選擇合適的數值類型
// 不好：使用 int64 存儲小數值
int64 sensor_id = 1; // 浪費空間
// 好：使用合適的類型
int32 sensor_id = 1; // 對大多數應用已經足夠
2. 批次傳輸
與其每次讀取就傳輸一次，不如累積幾筆資料一起傳：
message SensorBatch {
string device_id = 1;
repeated SensorReading readings = 2;
}
3. 壓縮大型資料
對於某些場景（如音訊資料、影像資料），可以在 protobuf 層面再加壓縮：
#include <ArduinoLZ77.h> // 輕量級壓縮庫
void compressAndSend(uint8_t* data, size_t length) {
uint8_t compressed[256];
size_t compressed_size = lz77_compress(data, length, compressed, sizeof(compressed));
if (compressed_size < length) {
// 壓縮有效，使用壓縮資料
sendToServer(compressed, compressed_size);
} else {
// 壓縮無效，使用原始資料
sendToServer(data, length);
}
}
除錯與測試技巧
開發過程中，除錯是不可避免的，這裡分享一些實用技巧。
序列化資料檢查
void printProtobufData(uint8_t* data, size_t length) {
Serial.print("Protobuf data (");
Serial.print(length);
Serial.print(" bytes): ");
for (size_t i = 0; i < length; i++) {
if (data[i] < 16) Serial.print("0");
Serial.print(data[i], HEX);
Serial.print(" ");
}
Serial.println();
}
反序列化測試
bool testSerialization() {
// 建立測試資料
SensorReading original = SensorReading_init_zero;
strcpy(original.device_id, "test_device");
original.temperature = 25.5f;
original.humidity = 60.0f;
// 序列化
uint8_t buffer[128];
pb_ostream_t ostream = pb_ostream_from_buffer(buffer, sizeof(buffer));
bool encode_status = pb_encode(&ostream, SensorReading_fields, &original);
if (!encode_status) {
Serial.println("Encoding failed");
return false;
}
// 反序列化
SensorReading decoded = SensorReading_init_zero;
pb_istream_t istream = pb_istream_from_buffer(buffer, ostream.bytes_written);
bool decode_status = pb_decode(&istream, SensorReading_fields, &decoded);
if (!decode_status) {
Serial.println("Decoding failed");
return false;
}
// 驗證資料
bool success = (strcmp(original.device_id, decoded.device_id) == 0) &&
(fabs(original.temperature - decoded.temperature) < 0.01f) &&
(fabs(original.humidity - decoded.humidity) < 0.01f);
if (success) {
Serial.println("Serialization test passed");
} else {
Serial.println("Serialization test failed");
}
return success;
}
記憶體洩漏檢查
void memoryLeakTest() {
uint32_t initial_free = ESP.getFreeHeap();
Serial.printf("Initial free heap: %d bytes\n", initial_free);
// 執行 1000 次序列化操作
for (int i = 0; i < 1000; i++) {
SensorReading reading = SensorReading_init_zero;
strcpy(reading.device_id, "test");
reading.temperature = i % 100;
uint8_t buffer[128];
pb_ostream_t stream = pb_ostream_from_buffer(buffer, sizeof(buffer));
pb_encode(&stream, SensorReading_fields, &reading);
if (i % 100 == 0) {
uint32_t current_free = ESP.getFreeHeap();
Serial.printf("Iteration %d, free heap: %d bytes\n", i, current_free);
}
}
uint32_t final_free = ESP.getFreeHeap();
Serial.printf("Final free heap: %d bytes\n", final_free);
Serial.printf("Memory change: %d bytes\n", (int32_t)final_free - (int32_t)initial_free);
}
實際專案案例：智能植栽監控系統
讓我分享一個完整的實際專案，展示如何在真實環境中應用這些技術。
專案背景
我幫朋友設計了一個智能植栽監控系統，用於管理他的小型有機農場。系統需要：
- 24/7 監控土壤濕度、光照、溫濕度
- 自動灌溉控制
- 手機 App 即時查看
- 低功耗運行（太陽能供電）
- 資料歷史記錄和分析
系統架構
感測器節點 → WiFi → MQTT Broker → 雲端伺服器 → 手機 App
↓
SD 卡備份
完整的 Proto Schema
syntax = "proto3";
// 感測器讀取資料
message SensorReading {
string node_id = 1;
int64 timestamp = 2;
string location = 3;
// 環境資料
float temperature = 10;
float humidity = 11;
int32 light_lux = 12;
// 土壤資料
float soil_moisture = 20;
float soil_temperature = 21;
float ph_level = 22;
// 系統狀態
float battery_voltage = 30;
int32 signal_strength = 31;
bool pump_active = 32;
// 錯誤和警告
repeated string warnings = 40;
}
// 控制命令
message ControlCommand {
string target_node = 1;
int64 timestamp = 2;
oneof command {
PumpControl pump = 10;
ConfigUpdate config = 11;
SystemCommand system = 12;
}
}
message PumpControl {
bool enable = 1;
int32 duration_seconds = 2;
}
message ConfigUpdate {
int32 reading_interval = 1;
float moisture_threshold = 2;
bool auto_irrigation = 3;
}
message SystemCommand {
enum Command {
RESTART = 0;
DEEP_SLEEP = 1;
FACTORY_RESET = 2;
UPDATE_FIRMWARE = 3;
}
Command command = 1;
}
Arduino 節點程式（簡化版）
#include <WiFi.h>
#include <PubSubClient.h>
#include <DHT.h>
#include <pb_encode.h>
#include <pb_decode.h>
#include "plant_monitor.pb.h"
// 硬體定義
#define DHT_PIN 4
#define DHT_TYPE DHT22
#define SOIL_MOISTURE_PIN A0
#define PUMP_RELAY_PIN 5
#define BATTERY_PIN A1
// 感測器物件
DHT dht(DHT_PIN, DHT_TYPE);
WiFiClient espClient;
PubSubClient mqtt(espClient);
// 設定參數
const char* wifi_ssid = "FarmWiFi";
const char* wifi_password = "your_password";
const char* mqtt_server = "farm-mqtt.example.com";
const char* node_id = "plant_node_001";
// 運行參數
int reading_interval = 300; // 5分鐘
float moisture_threshold = 30.0; // 30%
bool auto_irrigation = true;
void setup() {
Serial.begin(115200);
// 初始化硬體
dht.begin();
pinMode(PUMP_RELAY_PIN, OUTPUT);
digitalWrite(PUMP_RELAY_PIN, LOW);
// 連接網路
connectWiFi();
mqtt.setServer(mqtt_server, 1883);
mqtt.setCallback(onMqttMessage);
connectMQTT();
Serial.println("Plant monitoring node started");
}
void loop() {
if (!mqtt.connected()) {
connectMQTT();
}
mqtt.loop();
// 讀取感測器資料
SensorReading reading = readAllSensors();
// 檢查是否需要灌溉
if (auto_irrigation && reading.soil_moisture < moisture_threshold) {
activateIrrigation();
reading.pump_active = true;
}
// 發送資料
sendSensorData(reading);
// 深度睡眠節省電力
Serial.printf("Sleeping for %d seconds\n", reading_interval);
ESP.deepSleep(reading_interval * 1000000);
}
SensorReading readAllSensors() {
SensorReading reading = SensorReading_init_zero;
// 基本資訊
strcpy(reading.node_id, node_id);
reading.timestamp = getUnixTime();
strcpy(reading.location, "Greenhouse_A");
// 環境感測器
reading.temperature = dht.readTemperature();
reading.humidity = dht.readHumidity();
reading.light_lux = readLightSensor();
// 土壤感測器
reading.soil_moisture = readSoilMoisture();
reading.soil_temperature = readSoilTemperature();
reading.ph_level = readPHLevel();
// 系統狀態
reading.battery_voltage = readBatteryVoltage();
reading.signal_strength = WiFi.RSSI();
reading.pump_active = false;
// 檢查警告
checkWarnings(reading);
return reading;
}
void sendSensorData(const SensorReading& reading) {
uint8_t buffer[256];
pb_ostream_t stream = pb_ostream_from_buffer(buffer, sizeof(buffer));
bool status = pb_encode(&stream, SensorReading_fields, &reading);
if (!status) {
Serial.println("Failed to encode sensor data");
return;
}
// 發送到 MQTT
bool sent = mqtt.publish("farm/sensors/data", buffer, stream.bytes_written);
if (sent) {
Serial.printf("Sent %d bytes of sensor data\n", stream.bytes_written);
} else {
Serial.println("Failed to send sensor data");
// 保存到 SD 卡作為備份
saveToSDCard(buffer, stream.bytes_written);
}
}
void onMqttMessage(char* topic, byte* payload, unsigned int length) {
if (strcmp(topic, "farm/control/commands") == 0) {
// 解析控制命令
ControlCommand command = ControlCommand_init_zero;
pb_istream_t stream = pb_istream_from_buffer(payload, length);
if (pb_decode(&stream, ControlCommand_fields, &command)) {
processControlCommand(command);
}
}
}
void processControlCommand(const ControlCommand& command) {
if (strcmp(command.target_node, node_id) != 0) {
return; // 不是給這個節點的命令
}
switch (command.which_command) {
case ControlCommand_pump_tag:
if (command.command.pump.enable) {
activateIrrigation(command.command.pump.duration_seconds);
} else {
digitalWrite(PUMP_RELAY_PIN, LOW);
}
break;
case ControlCommand_config_tag:
// 更新配置
reading_interval = command.command.config.reading_interval;
moisture_threshold = command.command.config.moisture_threshold;
auto_irrigation = command.command.config.auto_irrigation;
Serial.println("Configuration updated");
break;
case ControlCommand_system_tag:
switch (command.command.system.command) {
case SystemCommand_Command_RESTART:
ESP.restart();
break;
case SystemCommand_Command_DEEP_SLEEP:
ESP.deepSleep(0);
break;
// ... 其他系統命令
}
break;
}
}
成果展示
這個系統運行了一年多，效果非常好：
資料傳輸效率：
- 平均每筆資料：45 bytes（protobuf）vs 180 bytes（JSON）
- 每天傳輸資料：約 288 筆 × 45 bytes = 12.96 KB
- 如果用 JSON：約 288 筆 × 180 bytes = 51.84 KB
- 節省了 75% 的頻寬
電池續航力：
- 使用 18650 鋰電池 + 太陽能板
- 持續運行時間：夏季無限，冬季可達 2 週（無陽光）
- 資料傳輸量減少直接延長了電池壽命
系統穩定性：
- 運行 365 天，只有 3 次因為網路問題丟失資料
- 本地 SD 卡備份機制確保資料不遺失
- 記憶體使用量穩定，無內存洩漏
開發工具與環境建置
讓我分享一套完整的開發工具鏈，讓你快速上手。
工具安裝
1. 安裝 nanopb
# 從 GitHub 下載
git clone https://github.com/nanopb/nanopb.git
cd nanopb
git submodule update --init
# 建置生成器
cd generator/proto
make
# 設定環境變數
export PATH=$PATH:/path/to/nanopb/generator
2. Arduino IDE 設定
在 Arduino IDE 中安裝所需的庫：
- PubSubClient（MQTT 客戶端）
- DHT sensor library
- ArduinoJson（如果需要 JSON 比較）
3. 建立專案結構
my_iot_project/
├── proto/
│ ├── sensor.proto
│ ├── sensor.options
│ └── generate.sh
├── arduino/
│ ├── main/
│ │ ├── main.ino
│ │ ├── sensor.pb.h
│ │ └── sensor.pb.c
│ └── libraries/
│ └── nanopb/
├── server/
│ ├── sensor_pb2.py
│ └── server.py
└── docs/
└── README.md
自動化腳本
建立 generate.sh 簡化開發流程：
#!/bin/bash
# proto/generate.sh
echo "Generating nanopb files..."
python /path/to/nanopb/generator/nanopb_generator.py sensor.proto
echo "Copying to Arduino project..."
cp sensor.pb.h sensor.pb.c ../arduino/main/
echo "Generating Python files..."
protoc --python_out=../server sensor.proto
echo "Done!"
VS Code 擴展配置
建立 .vscode/tasks.json：
{
"version": "2.0.0",
"tasks": [
{
"label": "Generate Protobuf",
"type": "shell",
"command": "./proto/generate.sh",
"group": "build",
"presentation": {
"echo": true,
"reveal": "always",
"focus": false,
"panel": "shared"
}
}
]
}
這樣你就可以用 Ctrl+Shift+P → "Tasks: Run Task" → "Generate Protobuf" 快速生成程式碼。
常見問題與解決方案
Q1: nanopb 編譯錯誤
問題：Arduino IDE 報告 nanopb 相關的編譯錯誤。
解決：
- 確認 nanopb 版本相容性（建議使用 0.4.x 版本）
- 檢查
.options檔案設定 - 確認生成的
.pb.h和.pb.c檔案在正確位置
// 在 .ino 檔案開頭加入
#define PB_ENABLE_MALLOC 0 // 禁用動態記憶體分配
#define PB_FIELD_32BIT 1 // 支援大型訊息
Q2: 記憶體不足
問題：ESP8266 出現記憶體不足，設備重啟。
解決：
- 減少緩衝區大小
- 使用更精簡的 protobuf schema
- 實作記憶體池管理
// 使用更小的緩衝區
#define PROTOBUF_BUFFER_SIZE 64 // 而不是 256
uint8_t buffer[PROTOBUF_BUFFER_SIZE];
Q3: 資料解析失敗
問題：伺服器端無法解析 Arduino 發送的 protobuf 資料。
解決：
- 確認兩端使用相同的
.proto檔案 - 檢查資料傳輸的 Content-Type
- 除錯序列化資料
// 除錯輸出
void debugProtobufData(uint8_t* data, size_t length) {
Serial.printf("Protobuf hex dump (%d bytes):\n", length);
for (int i = 0; i < length; i++) {
Serial.printf("%02X ", data[i]);
if ((i + 1) % 16 == 0) Serial.println();
}
Serial.println();
}
Q4: WiFi 連線問題
問題：在某些環境下 WiFi 連線不穩定。
解決：實作更強健的連線管理
void ensureWiFiConnection() {
int retry_count = 0;
const int max_retries = 10;
while (WiFi.status() != WL_CONNECTED && retry_count < max_retries) {
Serial.printf("WiFi connecting... (%d/%d)\n", retry_count + 1, max_retries);
WiFi.begin(ssid, password);
delay(5000);
retry_count++;
}
if (WiFi.status() != WL_CONNECTED) {
Serial.println("WiFi connection failed, entering deep sleep");
ESP.deepSleep(60 * 1000000); // 休眠 1 分鐘後重試
}
}
未來發展與技術趨勢
邊緣計算整合
隨著 ESP32-S3 等更強大的微控制器出現，我們可以在設備端進行更多資料處理：
message ProcessedData {
string device_id = 1;
int64 timestamp = 2;
// 原始資料的統計摘要
StatisticalSummary temperature_stats = 10;
StatisticalSummary humidity_stats = 11;
// 異常檢測結果
repeated AnomalyAlert anomalies = 20;
}
message StatisticalSummary {
float mean = 1;
float min = 2;
float max = 3;
float std_dev = 4;
int32 sample_count = 5;
}
機器學習推論
未來的 IoT 設備可能會整合 TinyML，在本地進行簡單的機器學習推論：
message MLPrediction {
string model_id = 1;
int64 timestamp = 2;
float confidence = 3;
oneof prediction {
PlantHealthPrediction plant_health = 10;
WeatherForecast weather = 11;
EquipmentFailure failure_risk = 12;
}
}
5G 和低軌衛星
隨著 5G 和低軌衛星網路的普及，更多偏遠地區的 IoT 設備可以接入網路。Protocol Buffers 的高效率將變得更加重要，特別是在衛星通信的高延遲環境下。
最後的話：小設備，大智慧
回想起那個讓我頭痛一個月的專案，如果當時就知道 nanopb 這個神器，該省多少時間啊！
Protocol Buffers 在 Arduino 和單晶片上的應用，讓我深刻體會到：限制往往是創新的催化劑。正是因為有了記憶體、頻寬、電力的限制，我們才被迫去尋找更高效的解決方案。
在 IoT 的世界裡，每個位元組都有它的價值。當你的設備需要運行數月甚至數年，當你的網路頻寬只有幾 KB/s，當你的記憶體只有幾 KB 時，選擇正確的技術就變得至關重要。
nanopb 不只是一個工具，它代表了一種思維方式：如何在有限的資源下做出無限的可能。
給新手的建議
如果你是第一次嘗試在 Arduino 上使用 Protocol Buffers，我的建議是：
- 從小開始：先用最簡單的 schema，確保整個流程跑得通
- 測量一切：記憶體使用量、傳輸時間、資料大小都要實際測量
- 保持耐心：除錯嵌入式系統比除錯伺服器程式困難得多
- 記錄經驗：把遇到的問題和解決方案記錄下來，下次會用到
展望未來
物聯網正在快速發展，邊緣計算、人工智慧、5G 通信等技術都在重塑這個領域。但無論技術如何進步，資源效率始終是嵌入式系統的核心議題。
Protocol Buffers 為我們提供了一個優雅的解決方案，讓小小的 Arduino 也能說一口流利的「高效語言」。
希望這篇文章能夠幫助你在自己的 IoT 專案中更好地應用 Protocol Buffers。如果你有任何問題或想要分享你的經驗，歡迎留言討論！
記住：在 IoT 的世界裡，小設備也能有大智慧。
相關資源
官方文檔
教學資源
開發工具
- Arduino IDE
- PlatformIO - 更強大的 Arduino 開發環境
- Buf - Protocol Buffers 開發工具
硬體建議
- 入門級：Arduino Uno + ESP8266 WiFi 模組
- 推薦級：ESP32 開發板（內建 WiFi/藍牙）
- 專業級：ESP32-S3 或 STM32 系列
標籤: #Arduino #ProtocolBuffers #nanopb #IoT #嵌入式系統 #ESP32 #感測器
本文最初發布於 HackMD @BASHCAT。
.jpg)
![Arduino 介紹 [Wiki]（圖 1）](http://zh.wikipedia.org/wiki/File:Arduino_tw_intro.jpg)
.jpg){kind=link}