做工业项目时，STM32 需要和 PLC 通信。Modbus RTU 是最常见的工业协议，libmodbus 是最流行的开源实现。但 libmodbus 是为 Linux 设计的，移植到 STM32 需要替换底层硬件操作。

为什么选 libmodbus

• 成熟稳定，社区活跃
• 支持 Modbus RTU/TCP
• API 设计清晰，易于使用
• 有丰富的示例代码

坑在于：libmodbus 依赖 POSIX API（open、read、write），STM32 裸机环境下没有这些。

移植思路

libmodbus 的架构分两层：

• 上层：Modbus 协议处理（CRC 计算、报文组装/解析）
• 底层：串口收发、定时器、延时

移植的关键是保留上层，替换底层。

第一步：简化源码

# 下载 libmodbus
git clone https://github.com/stephane/libmodbus.git

# 只保留核心文件
src/
├── modbus.c          # Modbus 协议核心
├── modbus.h          # 公共头文件
├── modbus-private.h  # 私有头文件
├── modbus-rtu.c      # RTU 模式实现
├── modbus-rtu.h      # RTU 模式头文件
└── config.h          # 配置文件

删除 TCP 相关文件（modbus-tcp.c、modbus-tcp.h），只保留 RTU 模式。

第二步：配置 config.h

// config.h
#define HAVE_STRLCPY 1
#define HAVE_STRERROR 1
#define HAVE_ARPA_INET_H 0  // STM32 没有
#define HAVE_NETINET_IN_H 0
#define HAVE_SYS_SOCKET_H 0
#define HAVE_WINSOCK2_H 0

坑在于：很多宏默认开启，STM32 环境下要手动关闭。

第三步：替换串口收发

libmodbus 的底层操作在 modbus-rtu.c 的 _step 函数里：

// 原始代码（Linux）
static ssize_t _step(modbus_t *ctx, uint8_t *msg, int msg_length)
{
    return write(ctx->s, msg, msg_length);
}

// 替换为 STM32 HAL
static ssize_t _step(modbus_t *ctx, uint8_t *msg, int msg_length)
{
    modbus_rtu_t *ctx_rtu = ctx->backend_data;
    HAL_StatusTypeDef status;

    // 发送
    status = HAL_UART_Transmit(&huart1, msg, msg_length, 1000);
    if (status != HAL_OK) {
        return -1;
    }

    // 接收
    status = HAL_UART_Receive(&huart1, msg, msg_length, 1000);
    if (status != HAL_OK) {
        return -1;
    }

    return msg_length;
}

第四步：替换延时函数

// 原始代码（Linux）
static void usleep_ms(int ms)
{
    usleep(ms * 1000);
}

// 替换为 STM32 HAL
static void usleep_ms(int ms)
{
    HAL_Delay(ms);
}

第五步：替换互斥锁（可选）

如果是单线程环境（裸机或 FreeRTOS 单任务），可以简化互斥锁：

// 简化为无操作
#define LOCK(ctx)
#define UNLOCK(ctx)

第六步：初始化和使用

#include "modbus.h"

int main(void)
{
    // 初始化硬件
    HAL_Init();
    SystemClock_Config();
    MX_USART1_UART_Init();

    // 创建 Modbus RTU 上下文
    modbus_t *ctx = modbus_new_rtu("/dev/ttyS0", 9600, 'N', 8, 1);
    if (ctx == NULL) {
        printf("无法创建 Modbus 上下文\n");
        return -1;
    }

    // 设置从站地址
    modbus_set_slave(ctx, 1);

    // 连接（RTU 模式不需要，但要调用）
    modbus_connect(ctx);

    // 读取保持寄存器（功能码 0x03）
    uint16_t tab_reg[10];
    int rc = modbus_read_registers(ctx, 0, 10, tab_reg);
    if (rc == 10) {
        printf("读取成功：");
        for (int i = 0; i < 10; i++) {
            printf("%d ", tab_reg[i]);
        }
        printf("\n");
    } else {
        printf("读取失败，错误码：%d\n", rc);
    }

    // 写入单个寄存器（功能码 0x06）
    rc = modbus_write_register(ctx, 0, 12345);
    if (rc == 1) {
        printf("写入成功\n");
    }

    // 关闭连接
    modbus_close(ctx);
    modbus_free(ctx);

    return 0;
}

常见问题

1. CRC 校验失败

// 检查波特率配置
modbus_t *ctx = modbus_new_rtu("/dev/ttyS0", 9600, 'N', 8, 1);
//                                    波特率    校验 数据位 停止位

坑在于：PLC 和 STM32 的波特率、校验位、数据位、停止位必须一致。

2. 超时问题

// 设置超时时间
struct timeval timeout;
timeout.tv_sec = 0;
timeout.tv_usec = 100000;  // 100ms
modbus_set_response_timeout(ctx, &timeout);

3. 从站地址不对

// 设置从站地址（要和 PLC 配置一致）
modbus_set_slave(ctx, 1);  // 从站地址 1

踩坑总结

1. libmodbus 依赖 POSIX API，STM32 需要替换底层操作
2. 核心替换：串口收发、延时函数、互斥锁
3. 保留上层协议处理（CRC、报文组装/解析）
4. 波特率、校验位、数据位、停止位必须和 PLC 一致
5. 从站地址要和 PLC 配置一致

从"libmodbus 不能用"到"STM32 和 PLC 通信成功"，只需要替换底层操作。坑在于：很多人被 POSIX API 吓退了，其实替换量不大。

部署一个 Python + Redis + Nginx 的服务，容器启动顺序搞错了。Redis 还没 ready，Python 服务就报 Connection refused。排查了半天，发现是 depends_on 只保证容器启动顺序，不保证服务就绪。

问题现象

# docker-compose.yml
services:
  web:
    build: .
    depends_on:
      - redis
    ports:
      - "8000:8000"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

docker-compose up -d

Python 服务启动后报错：

redis.exceptions.ConnectionError: Error 111 connecting to localhost:6379. Connection refused.

坑在于：depends_on 只保证 Redis 容器先启动，不保证 Redis 服务已经就绪。Redis 容器启动了，但 Redis 服务可能还在初始化。

解决方案

方案一：healthcheck + condition（推荐）

services:
  web:
    build: .
    depends_on:
      redis:
        condition: service_healthy
    ports:
      - "8000:8000"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

healthcheck 定义健康检查命令，condition: service_healthy 等待 Redis 健康后再启动 web。

方案二：启动脚本等待

# app.py
import redis
import time

def wait_for_redis(host, port, max_retries=30):
    """等待 Redis 就绪"""
    for i in range(max_retries):
        try:
            r = redis.Redis(host=host, port=port)
            r.ping()
            print("Redis 已就绪")
            return r
        except redis.ConnectionError:
            print(f"等待 Redis... ({i+1}/{max_retries})")
            time.sleep(1)
    raise Exception("Redis 连接超时")

r = wait_for_redis("redis", 6379)

坑在于：这个方案需要在应用代码里加等待逻辑，不够优雅。

方案三：wait-for-it 脚本

services:
  web:
    build: .
    depends_on:
      - redis
    command: >
      sh -c "wait-for-it -t 30 redis:6379 -- python app.py"
    ports:
      - "8000:8000"

# Dockerfile
FROM python:3.11-slim

# 安装 wait-for-it
RUN apt-get update && apt-get install -y wait-for-it && rm -rf /var/lib/apt/lists/*

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .

这个方案需要在镜像里安装 wait-for-it，增加镜像体积。

完整配置示例

Python + Redis + Nginx

services:
  web:
    build: .
    depends_on:
      redis:
        condition: service_healthy
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    ports:
      - "8000:8000"

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      web:
        condition: service_started

volumes:
  redis_data:

健康检查配置详解

healthcheck:
  test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
  interval: 10s      # 检查间隔
  timeout: 5s        # 超时时间
  retries: 3         # 重试次数
  start_period: 30s  # 启动等待时间

start_period 给容器启动的时间，这段时间内的失败不算在 retries 里。

网络配置

services:
  web:
    build: .
    networks:
      - app-network

  redis:
    image: redis:7-alpine
    networks:
      - app-network

networks:
  app-network:
    driver: bridge

同一个网络里的容器可以用服务名互相访问，比如 redis:6379。

常用命令

# 启动所有服务
docker-compose up -d

# 查看服务状态
docker-compose ps

# 查看日志
docker-compose logs -f web

# 重建并启动
docker-compose up -d --build

# 停止并清理
docker-compose down

# 停止并清理数据卷
docker-compose down -v

踩坑总结

1. depends_on 只保证容器启动顺序，不保证服务就绪
2. 用 healthcheck + condition: service_healthy 等待服务健康
3. start_period 给容器启动的时间，避免误判
4. 同一个网络里的容器可以用服务名互相访问
5. 生产环境用 docker-compose up -d --build 重建镜像

从"容器启动了但服务没就绪"到"服务健康后再启动依赖"，只需要配置 healthcheck + condition。坑在于：很多人只用 depends_on，不知道还有 condition 这个选项。

项目里到处都是重复的日志代码，每个函数开头写 logger.info("开始...")，结尾写 logger.info("结束")。后来用装饰器统一处理，代码简洁了很多。这篇文章记录一下装饰器的实战用法。

什么是闭包

装饰器的基础是闭包。简单说，闭包就是函数里定义的函数，能记住外层函数的变量。

def outer(x):
    def inner(y):
        return x + y  # inner 记住了 x 的值
    return inner

add5 = outer(5)
print(add5(3))  # 输出 8
print(add5(10))  # 输出 15

坑在于：outer(5) 执行完后，按理说 x 应该被销毁了。但 inner 函数记住了 x 的值，所以还能用。

最简单的装饰器

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} 执行完成")
        return result
    return wrapper

@log_decorator
def add(a, b):
    return a + b

add(1, 2)
# 输出：
# 调用 add
# add 执行完成

@log_decorator 等价于 add = log_decorator(add)。

带参数的装饰器

如果装饰器本身需要参数，要再包一层：

import time
from functools import wraps

def timer(threshold=None):
    def decorator(func):
        @wraps(func)  # 保留原函数的 __name__ 和 __doc__
        def wrapper(*args, **kwargs):
            start = time.time()
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            if threshold and elapsed > threshold:
                print(f"警告：{func.__name__} 耗时 {elapsed:.2f}s，超过阈值 {threshold}s")
            else:
                print(f"{func.__name__} 耗时 {elapsed:.2f}s")
            return result
        return wrapper
    return decorator

@timer(threshold=1)
def slow_function():
    time.sleep(2)

slow_function()
# 输出：警告：slow_function 耗时 2.00s，超过阈值 1s

坑在于：不加 @wraps(func)，装饰后的函数 __name__ 会变成 wrapper，调试时很困惑。

实战：日志装饰器

import logging
import time
from functools import wraps

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

def log_execution(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"开始执行 {func.__name__}，参数: args={args}, kwargs={kwargs}")
        start = time.time()
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"{func.__name__} 执行成功，耗时: {elapsed:.3f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"{func.__name__} 执行失败，耗时: {elapsed:.3f}s，错误: {e}")
            raise
    return wrapper

@log_execution
def process_data(data):
    time.sleep(0.5)
    return [x * 2 for x in data]

result = process_data([1, 2, 3])

输出：

2026-07-20 14:30:00 [INFO] 开始执行 process_data，参数: args=([1, 2, 3],), kwargs={}
2026-07-20 14:30:01 [INFO] process_data 执行成功，耗时: 0.501s

实战：重试装饰器

import time
from functools import wraps

def retry(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    print(f"第 {attempt + 1} 次失败：{e}，{delay}s 后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_retries=3, delay=2)
def unstable_api():
    import random
    if random.random() < 0.7:
        raise ConnectionError("API 连接失败")
    return "成功"

result = unstable_api()

这个装饰器在网络请求场景很有用，自动重试失败的请求。

实战：缓存装饰器

from functools import wraps

def cache(func):
    cached = {}
    @wraps(func)
    def wrapper(*args):
        if args in cached:
            print(f"命中缓存：{args}")
            return cached[args]
        result = func(*args)
        cached[args] = result
        return result
    return wrapper

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(10))  # 计算
print(fibonacci(10))  # 命中缓存

坑在于：这个简单缓存没有过期机制，如果数据量大会占用内存。Python 3.9+ 可以用 functools.lru_cache。

实战：权限检查装饰器

from functools import wraps

def require_permission(permission):
    def decorator(func):
        @wraps(func)
        def wrapper(user, *args, **kwargs):
            if permission not in user.get("permissions", []):
                raise PermissionError(f"用户 {user['name']} 没有 {permission} 权限")
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

@require_permission("admin")
def delete_user(user, user_id):
    print(f"删除用户 {user_id}")

admin = {"name": "admin", "permissions": ["admin", "read"]}
user = {"name": "guest", "permissions": ["read"]}

delete_user(admin, 123)  # 正常执行
# delete_user(user, 123)  # 抛出 PermissionError

踩坑总结

1. 装饰器本质是闭包，记住外层函数的变量
2. 用 @wraps(func) 保留原函数的元信息
3. 带参数的装饰器要再包一层
4. 装饰器顺序从下往上执行
5. Python 3.9+ 用 functools.lru_cache 做缓存，比自己实现更可靠

从 print() 调试到装饰器统一处理，只需要花半小时学习一次，但能省下无数重复代码。

做嵌入式项目时，I2C 通信总是不稳定，有时候能通有时候不通。排查了半天，最后发现是开漏输出没加上拉电阻。这个问题很常见，90% 的工程师都忽略了这一点。

问题现象

// 配置 GPIO 为开漏输出
GPIO_InitTypeDef GPIO_InitStruct = {0};
GPIO_InitStruct.Pin = GPIO_PIN_6;  // SCL
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_NOPULL;  // 没有上下拉
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

I2C 通信时好时坏，有时候能通，有时候直接卡死。用逻辑分析仪看波形，发现 SCL 线有时候拉不上去。

原因分析

说白了，就是开漏输出的特性决定的。

推挽输出（Push-Pull）：

• 可以主动输出高电平和低电平
• 内部有两个 MOS 管，一个拉高，一个拉低
• 不需要外部上下拉电阻

开漏输出（Open-Drain）：

• 只能主动输出低电平
• 高电平是靠外部上拉电阻拉上去的
• 如果没有上拉电阻，高电平就是浮空状态

坑在于：开漏输出的"高电平"不是 MCU 主动输出的，而是靠外部上拉电阻拉上去的。如果没有上拉，高电平就是浮空，电平不确定。

解决方案

方案一：配置内部上拉

// 配置 GPIO 为开漏输出 + 内部上拉
GPIO_InitTypeDef GPIO_InitStruct = {0};
GPIO_InitStruct.Pin = GPIO_PIN_6;  // SCL
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 内部上拉
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

坑在于：内部上拉电阻通常在 20kΩ-50kΩ，阻值比较大。对于 I2C 这种需要快速上升沿的场景，内部上拉可能不够。

方案二：外部上拉电阻（推荐）

// 在硬件上加外部上拉电阻
// SCL 线：4.7kΩ 上拉到 VCC
// SDA 线：4.7kΩ 上拉到 VCC

I2C 协议推荐使用 4.7kΩ 上拉电阻。如果总线上设备多或者线长，可以适当减小阻值（比如 2.2kΩ）。

方案三：根据场景选择

// 场景 1：普通 GPIO 控制 LED
// 推挽输出，不需要上下拉
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;
GPIO_InitStruct.Pull = GPIO_NOPULL;

// 场景 2：I2C 通信
// 开漏输出 + 外部上拉
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;
GPIO_InitStruct.Pull = GPIO_NOPULL;  // 用外部上拉

// 场景 3：按键输入
// 输入模式 + 内部上拉
GPIO_InitStruct.Mode = GPIO_MODE_INPUT;
GPIO_InitStruct.Pull = GPIO_PULLUP;

// 场景 4：UART TX
// 推挽输出，空闲状态保持高电平
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;
GPIO_InitStruct.Pull = GPIO_NOPULL;

常见场景配置

I2C 通信

// I2C GPIO 配置
GPIO_InitStruct.Pin = GPIO_PIN_6 | GPIO_PIN_7;  // SCL, SDA
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_OD;  // 开漏输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 内部上拉（建议用外部 4.7kΩ）
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOB, &GPIO_InitStruct);

SPI 通信

// SPI GPIO 配置
GPIO_InitStruct.Pin = GPIO_PIN_5 | GPIO_PIN_7;  // SCK, MOSI
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;  // 推挽输出
GPIO_InitStruct.Pull = GPIO_NOPULL;
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOA, &GPIO_InitStruct);

UART 通信

// UART TX 配置
GPIO_InitStruct.Pin = GPIO_PIN_9;  // TX
GPIO_InitStruct.Mode = GPIO_MODE_OUTPUT_PP;  // 推挽输出
GPIO_InitStruct.Pull = GPIO_PULLUP;  // 空闲状态高电平
GPIO_InitStruct.Speed = GPIO_SPEED_FREQ_HIGH;
HAL_GPIO_Init(GPIOA, &GPIO_InitStruct);

踩坑总结

1. 开漏输出只能主动拉低，高电平靠外部上拉
2. I2C 必须用开漏输出 + 上拉电阻
3. SPI、UART 用推挽输出，不需要上下拉
4. 内部上拉电阻阻值大，高速通信建议用外部上拉
5. 记住口诀："推挽上下拉是白搭，开漏不拉就抓瞎"

这个问题的坑在于：开漏输出的高电平是靠上拉电阻实现的，没有上拉就是浮空。理解了这个原理，配置就很清晰了。

凌晨 2 点收到告警，服务挂了。SSH 上去一看，No space left on device。df 一看，根分区 100% 了。但 du 一算，明明只用了 20G，分区有 40G。坑在于：有已删除的文件还占着空间。

问题现象

df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/vda1        40G   40G     0 100% /

du -sh /*
# 加起来只有 20G

df 说用了 40G，du 说只用了 20G。差的 20G 去哪了？

原因分析

说白了，就是有文件被删除了，但进程还持有文件句柄。Linux 的机制是：文件被删除后，只要还有进程在用它，磁盘空间就不会释放。

坑在于：du 只统计当前存在的文件，不统计已删除但还被占用的文件。

排查过程

第一步：找出已删除但未释放的文件

# 查找被删除但还被占用的文件
lsof +L1

# 输出类似：
# COMMAND    PID USER   FD   TYPE DEVICE SIZE/NODE NODE NAME
# java     12345 root    1w  REG  253,1 21474836480  1234567 /var/log/app.log (deleted)

lsof +L1 查找所有已删除但还被占用的文件（links < 1）。输出显示 java 进程占用了一个 20G 的已删除文件。

第二步：确认文件大小

# 查看被删除文件的大小
ls -lh /proc/12345/fd/1

# 或者
ls -l /proc/12345/fd/ | grep deleted

# 输出：
# l-wx------ 1 root root 64 Jul 20 02:00 1 -> /var/log/app.log (deleted)

确认了，/var/log/app.log 被删除了，但 java 进程还持有它的句柄，占着 20G 空间。

第三步：释放空间

有两个方案：

方案一：重启进程（推荐）

# 重启 java 进程，释放文件句柄
systemctl restart java-app

重启后，已删除的文件句柄被释放，磁盘空间回来了。

方案二：不重启，清空文件内容

# 找到被删除文件的文件描述符
ls -l /proc/12345/fd/ | grep deleted
# 输出：1 -> /var/log/app.log (deleted)

# 清空文件内容（不删除文件，只清空）
> /proc/12345/fd/1

这个方法不需要重启进程，但坑在于：只能清空当前内容，进程继续写入还是会占用空间。

验证修复

# 重启后检查
df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/vda1        40G   20G   18G  53% /

# 确认没有已删除未释放的文件
lsof +L1
# 无输出

空间回来了，从 100% 降到 53%。

预防措施

1. 配置日志轮转

# /etc/logrotate.d/app
/var/log/app.log {
    daily
    rotate 7
    compress
    delaycompress
    missingok
    notifempty
    copytruncate
}

copytruncate 会先复制再清空，避免进程持有已删除文件的句柄。

2. 监控磁盘使用率

# 添加 crontab 定期检查
*/5 * * * * df -h / | awk 'NR==2{print $5}' | sed 's/%//' | awk '$1>80{print "磁盘使用率超过80%: "$1"%"}' | mail -s "磁盘告警" admin@example.com

3. 清理大文件

# 找出大于 100M 的文件
find / -type f -size +100M -exec ls -lh {} \; 2>/dev/null

# 找出大于 7 天的日志文件
find /var/log -type f -name "*.log" -mtime +7 -exec ls -lh {} \;

踩坑总结

1. df 和 du 结果不一致，通常是有已删除未释放的文件
2. lsof +L1 可以找出已删除但还被占用的文件
3. 重启进程可以释放文件句柄
4. 配置日志轮转（logrotate）可以预防此类问题
5. 定期监控磁盘使用率，设置告警阈值

这个问题的坑在于：df 说满了，du 说没满。理解了 Linux 的文件删除机制，排查就很清晰了。