FPGAZynq工业通信ModbusEtherCATMQTT边缘计算FreeRTOSOpenAMPPetaLinux
Zynq 实战 28｜项目实战三：工业通信网关（Modbus + EtherCAT + MQTT 边缘计算）
Zynq 实战 28｜项目实战三：工业通信网关（Modbus + EtherCAT + MQTT 边缘计算）
这是《Zynq FPGA 嵌入式系统设计实战》系列的第 28 篇。 板子：Pynq-Z2（XC7Z020-1CLG400C）。工具链：Vivado / Vitis / PetaLinux 2023.2。 上一篇：《Zynq 实战 27｜项目实战二：机器视觉平台》
0. 这一篇要解决什么问题
工业设备不仅仅要处理数据，还要把数据从设备侧搬到云端，同时响应来自多种协议的设备。这就是工业通信网关的核心任务。
这一篇要做的系统：
现场设备（EtherCAT 从站 / Modbus RTU 设备）
↓ 东西向（设备侧）
Zynq PS (Linux + FreeRTOS) + PL（时间戳 IP）
↓ 南北向（上云）
MQTT → AWS IoT Core / 阿里云 IoT具体技术栈：
- Modbus RTU（RS-485）+ Modbus TCP：libmodbus 库，PetaLinux 上完整实现
- EtherCAT：SOEM 移植，GEM0 独占，1ms 循环时间，FreeRTOS 处理实时部分
- PL 硬件时间戳：IEEE 1588 精度，给 EtherCAT 帧打时间戳
- MQTT 上云：mosquitto，JSON 格式，QoS 2，mbedTLS 加密
- 边缘计算：Z-score 异常检测，C 实现，在网关上做初步过滤
本文不覆盖：OPC UA（工业 4.0 标准协议，体量太大单独一篇）；PROFINET（需要专用 PHY）。
1. 系统架构总览
2. Modbus RTU（RS-485）实现
2.1 硬件电路：Pynq-Z2 + RS-485 收发器
Pynq-Z2 板上有两路 UART（MIO 14/15 和 MIO 48/49），通过外接 RS-485 收发器（如 MAX485、SN75176）转换为差分总线信号。
Pynq-Z2 MIO14 (TX) ─────┐
Pynq-Z2 MIO15 (RX) ─────┤ MAX485 芯片
Pynq-Z2 MIO16 (GPIO) ───┘ (DE/RE 控制端)
↓ RS-485 差分总线（A/B 线）
[终端电阻 120Ω]─────────── Modbus RTU 从站 × N ──────────[终端电阻 120Ω]关键参数：
- 波特率：19200 或 115200（工业标准多为 9600-19200）
- RS-485 总线长度：19200 bps 下最长 1200m，115200 bps 下约 120m
- 从站最多：理论 247 个（Modbus 协议地址范围 1-247）
- 实际推荐：单总线不超过 32 个节点（MAX485 驱动能力限制）
🚧 避坑：RS-485 总线两端必须各接一个 120Ω 终端电阻（差分线 A/B 之间）。没有终端电阻的总线在长距离或高速率下会出现反射，表现为 Modbus CRC 错误率飙高（从 <0.1% 升到 >10%）。测试短距离（<1m）时终端电阻可以不加，但生产环境一定要加。另外，MAX485 的 DE/RE（驱动使能/接收使能）引脚要配合 UART TX 方向切换——发送时拉高 DE（启用发送），接收时拉低（启用接收），可以用 Pynq-Z2 的 GPIO MIO16 控制。
2.2 PetaLinux 设备树：UART + RS-485 模式
/* system-user.dtsi */
&uart0 {
status = "okay";
/* 使能 RS-485 半双工模式（内核 rs485 框架） */
linux,rs485-enabled-at-boot-time;
rs485-rts-active-high; /* DE 引脚高电平有效 */
rts-gpios = <&gpio0 16 GPIO_ACTIVE_HIGH>; /* MIO16 → MAX485 DE/RE */
};2.3 libmodbus：读取 Modbus RTU 保持寄存器
/*
* modbus_rtu_master.c — Modbus RTU 主站，读取从站保持寄存器
*
* 编译：arm-linux-gnueabihf-gcc -O2 -o modbus_rtu_master modbus_rtu_master.c -lmodbus
*
* 依赖：libmodbus（PetaLinux rootfs 通过 packagegroup-base 包含，
* 或手动加到 recipes-support/libmodbus/）
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <modbus.h>
#define UART_DEV "/dev/ttyUL0" /* MIO UART0，对应 RS-485 总线 */
#define BAUD_RATE 19200
#define SLAVE_ADDR 1 /* Modbus 从站地址（1-247） */
#define REG_START 0 /* 保持寄存器起始地址 */
#define REG_COUNT 10 /* 读取 10 个寄存器 */
int main(void) {
modbus_t *ctx;
uint16_t regs[REG_COUNT];
int ret;
/* ── 创建 Modbus RTU 上下文 ── */
ctx = modbus_new_rtu(UART_DEV, BAUD_RATE,
'N', /* 无奇偶校验 */
8, /* 数据位 */
1); /* 停止位 */
if (!ctx) {
fprintf(stderr, "modbus_new_rtu 失败: %s\n", modbus_strerror(errno));
return 1;
}
/* ── 设置从站地址 ── */
modbus_set_slave(ctx, SLAVE_ADDR);
/* ── 设置响应超时：500ms（工业环境建议 200-1000ms） ── */
modbus_set_response_timeout(ctx, 0, 500000); /* 秒, 微秒 */
/* ── 连接（打开串口） ── */
if (modbus_connect(ctx) < 0) {
fprintf(stderr, "modbus_connect 失败: %s\n", modbus_strerror(errno));
modbus_free(ctx);
return 1;
}
/* ── 读取保持寄存器（功能码 0x03） ── */
ret = modbus_read_registers(ctx, REG_START, REG_COUNT, regs);
if (ret < 0) {
fprintf(stderr, "读取失败: %s\n", modbus_strerror(errno));
} else {
printf("从站 %d，寄存器 %d-%d:\n", SLAVE_ADDR, REG_START, REG_START+REG_COUNT-1);
for (int i = 0; i < REG_COUNT; i++)
printf(" [%04d] = 0x%04X (%u)\n", REG_START+i, regs[i], regs[i]);
}
modbus_close(ctx);
modbus_free(ctx);
return ret < 0 ? 1 : 0;
}2.4 Modbus TCP 服务端（让 SCADA 来连）
/*
* modbus_tcp_server.c — Modbus TCP 服务端（网关角色）
* SCADA 系统通过 Modbus TCP 连接到网关，网关透明代理到 RTU 总线
*
* 编译：arm-linux-gnueabihf-gcc -O2 -o modbus_tcp_server modbus_tcp_server.c -lmodbus -lpthread
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <pthread.h>
#include <modbus.h>
#include <sys/socket.h>
#define TCP_PORT 502 /* Modbus TCP 标准端口 */
#define MAX_REGS 100 /* 模拟寄存器区大小 */
static uint16_t holding_regs[MAX_REGS]; /* 共享数据区 */
static pthread_mutex_t regs_mutex = PTHREAD_MUTEX_INITIALIZER;
/* 后台线程：周期性从 RTU 总线采集数据更新 holding_regs */
static void *rtu_poll_thread(void *arg) {
(void)arg;
/* 实际工程里在这里循环调用 modbus_read_registers()，写入 holding_regs */
/* 本示例用随机数模拟 */
while (1) {
pthread_mutex_lock(®s_mutex);
for (int i = 0; i < MAX_REGS; i++)
holding_regs[i] = (uint16_t)(rand() % 65536);
pthread_mutex_unlock(®s_mutex);
usleep(10000); /* 10ms 轮询一次 */
}
return NULL;
}
int main(void) {
modbus_t *ctx;
modbus_mapping_t *mb_mapping;
int server_socket, client_socket;
uint8_t query[MODBUS_TCP_MAX_ADU_LENGTH];
int rc;
pthread_t poll_tid;
/* 启动 RTU 采集线程 */
pthread_create(&poll_tid, NULL, rtu_poll_thread, NULL);
/* 创建 Modbus TCP 上下文 */
ctx = modbus_new_tcp("0.0.0.0", TCP_PORT);
if (!ctx) { perror("modbus_new_tcp"); return 1; }
/* 分配寄存器映射（支持线圈、输入、保持、输入寄存器） */
mb_mapping = modbus_mapping_new(0, 0, MAX_REGS, MAX_REGS);
if (!mb_mapping) { perror("modbus_mapping_new"); return 1; }
server_socket = modbus_tcp_listen(ctx, 1);
printf("Modbus TCP 服务端监听 0.0.0.0:%d\n", TCP_PORT);
while (1) {
client_socket = modbus_tcp_accept(ctx, &server_socket);
if (client_socket < 0) continue;
printf("SCADA 客户端已连接\n");
while (1) {
/* 每次处理一个请求前，把最新数据同步到 mb_mapping */
pthread_mutex_lock(®s_mutex);
memcpy(mb_mapping->tab_registers, holding_regs,
MAX_REGS * sizeof(uint16_t));
pthread_mutex_unlock(®s_mutex);
rc = modbus_receive(ctx, query);
if (rc <= 0) break; /* 客户端断开 */
modbus_reply(ctx, query, rc, mb_mapping);
}
printf("SCADA 客户端断开\n");
close(client_socket);
}
modbus_mapping_free(mb_mapping);
modbus_free(ctx);
return 0;
}3. EtherCAT：SOEM 移植与 1ms 实时循环
3.1 SOEM 概述
SOEM（Simple Open EtherCAT Master）是开源的 EtherCAT 主站实现，通过直接操作以太网帧实现 EtherCAT 通信，不依赖内核网络栈。
关键特性：
- 直接使用 RAW socket（
AF_PACKET）访问以太网帧 - 需要独占网口（不能与 TCP/IP 共享）
- 支持 CoE（CANopen over EtherCAT）、FoE（File over EtherCAT）
- 1ms 循环时间需要实时 Linux 或 FreeRTOS 支持
3.2 SOEM 移植到 PetaLinux
# 1. 在 PetaLinux 项目里添加 SOEM recipe
mkdir -p project-spec/meta-user/recipes-support/soem
cat > project-spec/meta-user/recipes-support/soem/soem_git.bb << 'EOF'
SUMMARY = "Simple Open EtherCAT Master"
LICENSE = "GPL-2.0-only"
LIC_FILES_CHKSUM = "file://LICENSE;md5=..."
SRC_URI = "git://github.com/OpenEtherCATsociety/SOEM.git;protocol=https;branch=master"
SRCREV = "v1.4.0"
S = "${WORKDIR}/git"
inherit cmake
EXTRA_OECMAKE = "-DCMAKE_BUILD_TYPE=Release"
do_install() {
install -d ${D}${libdir}
install -m 0755 ${B}/libsoem.a ${D}${libdir}/
install -d ${D}${includedir}/soem
install -m 0644 ${S}/soem/*.h ${D}${includedir}/soem/
}
EOF
# 2. 在 project-spec/meta-user/conf/layer.conf 里添加 soem 到 IMAGE_INSTALL
# IMAGE_INSTALL += "soem"3.3 EtherCAT 主站实现（SOEM）
/*
* ethercat_master.c — 基于 SOEM 的 EtherCAT 主站
* 运行在 Linux CPU0 上，或移植到 FreeRTOS CPU1（推荐后者以获得 1ms 确定性）
*
* 编译：arm-linux-gnueabihf-gcc -O2 -o ethercat_master ethercat_master.c \
* -lsoem -lpthread -lrt
*
* 注意：需要 root 权限（RAW socket 需要）
* GEM0 必须独占（不能有 IP 地址，不能跑 TCP/IP 协议栈）
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <time.h>
#include <errno.h>
#include "soem/ethercat.h"
#define ETHERCAT_IFACE "eth0" /* GEM0 网口（独占，不配 IP） */
#define CYCLE_NS (1000000L) /* 1ms 循环时间（单位：纳秒） */
#define EC_TIMEOUTMON 500 /* 从站监控超时：500ms */
/* 进程数据输出缓冲（PDO 数据） */
static char io_map[4096];
static volatile int wkc; /* working counter */
static volatile int do_run = 1;
/* 环形缓冲区：EtherCAT 采集数据 → MQTT 上报线程 */
#define RING_SIZE 1024
typedef struct {
uint64_t timestamp_ns; /* IEEE 1588 时间戳（来自 PL IP） */
uint32_t slave_id;
uint16_t data[8]; /* 每从站 8 个 PDO 字 */
} ec_sample_t;
static ec_sample_t ring_buf[RING_SIZE];
static volatile int ring_head = 0, ring_tail = 0;
static pthread_mutex_t ring_mutex = PTHREAD_MUTEX_INITIALIZER;
/* 读取 PL 硬件时间戳（通过 /dev/mem 或 UIO 访问 IEEE 1588 IP） */
static uint64_t read_hw_timestamp(void) {
/* TODO: 映射到 AXI-Lite 基地址后读取 64-bit 时间戳寄存器 */
/* 这里用 clock_gettime 临时替代（精度 ~1μs vs 硬件 ~10ns） */
struct timespec ts;
clock_gettime(CLOCK_REALTIME, &ts);
return (uint64_t)ts.tv_sec * 1000000000ULL + ts.tv_nsec;
}
/* 实时采集线程（理想情况跑在 FreeRTOS CPU1 上） */
static void *ec_cyclic_task(void *arg) {
struct timespec ts;
(void)arg;
/* 设置线程实时优先级（SCHED_FIFO，优先级 80） */
struct sched_param sp = { .sched_priority = 80 };
if (pthread_setschedparam(pthread_self(), SCHED_FIFO, &sp) != 0)
fprintf(stderr, "警告: 无法设置实时优先级（需要 root 或 CAP_SYS_NICE）\n");
/* 绑定到 CPU1（避免和 MQTT 线程竞争 CPU0） */
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(1, &cpuset);
pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset);
clock_gettime(CLOCK_MONOTONIC, &ts);
while (do_run) {
/* ── 等待下一个 1ms 周期起点 ── */
ts.tv_nsec += CYCLE_NS;
while (ts.tv_nsec >= 1000000000L) {
ts.tv_nsec -= 1000000000L;
ts.tv_sec++;
}
clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME, &ts, NULL);
/* ── 发送 EtherCAT 帧，接收响应 ── */
ec_send_processdata();
wkc = ec_receive_processdata(EC_TIMEOUTRET);
if (wkc >= ec_slave[0].outputs) {
/* 把所有从站的 PDO 数据放入环形缓冲区 */
uint64_t hw_ts = read_hw_timestamp();
pthread_mutex_lock(&ring_mutex);
int next_head = (ring_head + 1) % RING_SIZE;
if (next_head != ring_tail) { /* 缓冲区未满 */
ec_sample_t *s = &ring_buf[ring_head];
s->timestamp_ns = hw_ts;
s->slave_id = 0; /* 简化：只记录从站 1 的数据 */
/* ec_slave[1].inputs 指向 io_map 中从站 1 的输入 PDO */
memcpy(s->data, ec_slave[1].inputs, sizeof(s->data));
ring_head = next_head;
}
pthread_mutex_unlock(&ring_mutex);
}
}
return NULL;
}
int main(void) {
pthread_t ec_tid;
/* ── 初始化 SOEM，绑定到 GEM0 ── */
if (!ec_init(ETHERCAT_IFACE)) {
fprintf(stderr, "ec_init(%s) 失败——确认 GEM0 没有 IP，且有 root 权限\n",
ETHERCAT_IFACE);
return 1;
}
printf("GEM0 EtherCAT 初始化成功\n");
/* ── 扫描从站 ── */
if (ec_config_init(FALSE) <= 0) {
fprintf(stderr, "未发现 EtherCAT 从站\n");
ec_close();
return 1;
}
printf("发现 %d 个从站\n", ec_slavecount);
/* ── 配置 PDO 映射，分配 io_map ── */
ec_config_map(&io_map);
ec_configdc(); /* 配置分布式时钟（DC） */
/* ── 切换到 OP 状态 ── */
ec_slave[0].state = EC_STATE_OPERATIONAL;
ec_writestate(0);
ec_statecheck(0, EC_STATE_OPERATIONAL, EC_TIMEOUTSTATE * 4);
if (ec_slave[0].state != EC_STATE_OPERATIONAL) {
fprintf(stderr, "从站未能进入 OP 状态\n");
ec_close();
return 1;
}
printf("所有从站进入 OP 状态，启动 1ms 循环\n");
/* ── 启动实时采集线程 ── */
pthread_create(&ec_tid, NULL, ec_cyclic_task, NULL);
/* 主线程等待（实际工程里在这里处理 SIGINT 信号） */
pause();
do_run = 0;
pthread_join(ec_tid, NULL);
ec_close();
return 0;
}🚧 避坑：EtherCAT 必须独占网口（GEM0），绝对不能同时跑 Linux TCP/IP 协议栈。具体操作是：在 Linux 启动后，不要给 eth0 配 IP 地址（不要在
/etc/network/interfaces里加 eth0 条目），也不要跑 DHCP 客户端。否则内核网络栈和 SOEM 会抢着收同一张网卡上的以太网帧，SOEM 会经常漏帧，循环时间从 1ms 飙到 10-100ms，完全不可用。Modbus TCP 和 MQTT 通过 GEM1（eth1）出网。
4. PL：IEEE 1588 硬件时间戳 IP
IEEE 1588（PTP，精确时间协议）硬件时间戳的核心是：在以太网帧到达 PHY 的精确时刻打上时间戳，精度可达 ±10ns。
4.1 时间戳 IP 寄存器定义
AXI-Lite 基地址：0x43D00000（在 Vivado Address Editor 分配）
偏移 宽度 名称 说明
0x00 32 TIMESTAMP_LO 当前时间戳低 32 位（纳秒，0-999999999）
0x04 32 TIMESTAMP_HI 当前时间戳高 32 位（秒）
0x08 32 LATCH_LO 帧到达时刻锁存值低 32 位
0x0C 32 LATCH_HI 帧到达时刻锁存值高 32 位
0x10 32 CTRL [0]=enable [1]=latch_irq_en [2]=pps_out_en
0x14 32 STATUS [0]=latch_valid [1]=pps_pulse4.2 时间戳 IP Verilog（核心模块）
// ptp_timestamp.v — 简化的 IEEE 1588 硬件时间戳 IP
// 功能：维护一个纳秒计数器，在外部触发（以太网帧 SFD 检测）时锁存时间
// 精度：受 PL 时钟精度限制，200 MHz 时钟分辨率 = 5ns
// 同步：通过 AXI-Lite 提供 PPS (Pulse Per Second) 对时接口
`timescale 1ns / 1ps
module ptp_timestamp #(
parameter CLK_FREQ_HZ = 200_000_000 // PL 时钟频率
) (
input wire aclk,
input wire aresetn,
// AXI-Lite 从机接口（寄存器访问）
input wire [4:0] s_axi_awaddr,
input wire s_axi_awvalid,
output wire s_axi_awready,
input wire [31:0] s_axi_wdata,
input wire s_axi_wvalid,
output wire s_axi_wready,
output wire [1:0] s_axi_bresp,
output wire s_axi_bvalid,
input wire s_axi_bready,
input wire [4:0] s_axi_araddr,
input wire s_axi_arvalid,
output wire s_axi_arready,
output reg [31:0] s_axi_rdata,
output wire [1:0] s_axi_rresp,
output wire s_axi_rvalid,
input wire s_axi_rready,
// 外部触发（来自以太网 MAC 的 SFD 检测信号）
input wire frame_sfd_in, // EtherCAT 帧开始标志
output wire irq_out // 锁存完成中断
);
// 内部时间计数器
reg [31:0] ns_counter; // 纳秒计数（0 to 999_999_999）
reg [31:0] sec_counter; // 秒计数
// 锁存寄存器
reg [31:0] latch_ns;
reg [31:0] latch_sec;
reg latch_valid;
// 控制寄存器
reg enable;
reg latch_irq_en;
localparam NS_PER_CLK = 1_000_000_000 / CLK_FREQ_HZ; // = 5 ns @ 200 MHz
// ── 时间计数器 ──
always @(posedge aclk) begin
if (!aresetn) begin
ns_counter <= 0;
sec_counter <= 0;
end else if (enable) begin
if (ns_counter >= (1_000_000_000 - NS_PER_CLK)) begin
ns_counter <= 0;
sec_counter <= sec_counter + 1;
end else begin
ns_counter <= ns_counter + NS_PER_CLK;
end
end
end
// ── 帧到达时间锁存 ──
always @(posedge aclk) begin
if (!aresetn) begin
latch_ns <= 0;
latch_sec <= 0;
latch_valid <= 0;
end else if (frame_sfd_in && enable) begin
latch_ns <= ns_counter;
latch_sec <= sec_counter;
latch_valid <= 1;
end else if (/* 读取锁存寄存器后自动清除 */ 0) begin
latch_valid <= 0;
end
end
assign irq_out = latch_valid && latch_irq_en;
// ── AXI-Lite 读逻辑（简化，只实现读路径） ──
assign s_axi_arready = 1;
assign s_axi_rresp = 2'b00;
assign s_axi_rvalid = s_axi_arvalid;
assign s_axi_awready = 1;
assign s_axi_wready = 1;
assign s_axi_bresp = 2'b00;
assign s_axi_bvalid = s_axi_wvalid;
always @(posedge aclk) begin
case (s_axi_araddr[4:2])
3'd0: s_axi_rdata <= ns_counter;
3'd1: s_axi_rdata <= sec_counter;
3'd2: s_axi_rdata <= latch_ns;
3'd3: s_axi_rdata <= latch_sec;
3'd4: s_axi_rdata <= {29'b0, latch_irq_en, enable, latch_valid};
default: s_axi_rdata <= 32'hDEADBEEF;
endcase
end
// ── AXI-Lite 写逻辑 ──
always @(posedge aclk) begin
if (!aresetn) begin
enable <= 0; latch_irq_en <= 0;
end else if (s_axi_wvalid && s_axi_awvalid) begin
case (s_axi_awaddr[4:2])
3'd4: begin
enable <= s_axi_wdata[0];
latch_irq_en <= s_axi_wdata[1];
end
endcase
end
end
endmodule5. MQTT 上云实现
5.1 PetaLinux rootfs 添加 mosquitto
# petalinux-config -c rootfs
# Filesystem Packages → libs → mosquitto → [*] mosquitto
# Filesystem Packages → libs → mosquitto → [*] mosquitto-dev
# Filesystem Packages → libs → openssl → [*] openssl (for TLS)🚧 避坑：MQTT TLS（
mqtts://）需要 mbedTLS 或 OpenSSL。PetaLinux 默认不包含 OpenSSL，需要在 rootfs menuconfig 里手动选上openssl。如果遗漏，mosquitto 连接 AWS IoT Core 时会报SSL handshake failed，排查时可以先用mosquitto_pub -h broker -p 1883（非 TLS 端口）验证网络通，再换到 8883 端口 + 证书。
5.2 MQTT 上报程序
/*
* mqtt_uploader.c — 从环形缓冲区取 EtherCAT 数据，上报到 AWS IoT Core
*
* 编译：arm-linux-gnueabihf-gcc -O2 -o mqtt_uploader mqtt_uploader.c \
* -lmosquitto -lpthread
*
* 依赖：mosquitto 1.6+ 客户端库，AWS IoT Core 证书文件
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <mosquitto.h>
#include <time.h>
/* AWS IoT Core 连接参数 */
#define MQTT_HOST "xxxxxxxx.iot.ap-northeast-1.amazonaws.com"
#define MQTT_PORT 8883
#define MQTT_TOPIC "gateway/pynqz2/sensors"
#define CLIENT_ID "zynq-gateway-01"
/* TLS 证书路径（scp 到板子的 /etc/iot/ 目录） */
#define CA_CERT "/etc/iot/root-CA.crt"
#define CLIENT_CERT "/etc/iot/device.pem.crt"
#define CLIENT_KEY "/etc/iot/device.private.key"
/* 上报间隔：500ms（EtherCAT 1ms 采样，每 500 个点上报一次均值） */
#define REPORT_INTERVAL_MS 500
static int mqtt_connected = 0;
static void on_connect(struct mosquitto *mosq, void *obj, int rc) {
(void)obj;
if (rc == MOSQ_ERR_SUCCESS) {
printf("MQTT 已连接到 AWS IoT Core\n");
mqtt_connected = 1;
} else {
fprintf(stderr, "MQTT 连接失败: %s\n", mosquitto_connack_string(rc));
}
}
static void on_publish(struct mosquitto *mosq, void *obj, int mid) {
(void)mosq; (void)obj;
/* mid 是消息 ID，QoS 2 时用于追踪 PUBCOMP */
}
int main(void) {
struct mosquitto *mosq;
char payload[512];
int rc;
mosquitto_lib_init();
mosq = mosquitto_new(CLIENT_ID, true, NULL);
if (!mosq) { perror("mosquitto_new"); return 1; }
/* TLS 配置 */
rc = mosquitto_tls_set(mosq, CA_CERT, NULL, CLIENT_CERT, CLIENT_KEY, NULL);
if (rc != MOSQ_ERR_SUCCESS) {
fprintf(stderr, "TLS 配置失败: %s\n", mosquitto_strerror(rc));
return 1;
}
mosquitto_tls_insecure_set(mosq, false); /* 验证服务器证书 */
/* 设置回调 */
mosquitto_connect_callback_set(mosq, on_connect);
mosquitto_publish_callback_set(mosq, on_publish);
/* 异步连接 */
rc = mosquitto_connect_async(mosq, MQTT_HOST, MQTT_PORT, 60);
if (rc != MOSQ_ERR_SUCCESS) {
fprintf(stderr, "mosquitto_connect_async: %s\n", mosquitto_strerror(rc));
return 1;
}
mosquitto_loop_start(mosq); /* 后台线程处理 MQTT 网络 IO */
/* 等待连接建立 */
int wait = 0;
while (!mqtt_connected && wait++ < 100) usleep(100000);
if (!mqtt_connected) { fprintf(stderr, "连接超时\n"); return 1; }
/* ── 主循环：每 500ms 聚合一次数据并上报 ── */
while (1) {
usleep(REPORT_INTERVAL_MS * 1000);
/* 这里从 EtherCAT 环形缓冲区取最新 500 个采样，计算均值 */
/* 简化：直接用模拟数据 */
double avg_val = 42.0 + (rand() % 1000) / 100.0;
/* JSON 格式上报 */
struct timespec ts;
clock_gettime(CLOCK_REALTIME, &ts);
int n = snprintf(payload, sizeof(payload),
"{"
"\"device_id\":\"%s\","
"\"timestamp\":%ld.%09ld,"
"\"sensors\":{"
"\"channel_0\":%.3f,"
"\"channel_1\":%.3f"
"}"
"}",
CLIENT_ID,
(long)ts.tv_sec, (long)ts.tv_nsec,
avg_val,
avg_val * 0.95
);
/* QoS 2：保证恰好一次送达 */
rc = mosquitto_publish(mosq, NULL, MQTT_TOPIC, n, payload, 2, false);
if (rc != MOSQ_ERR_SUCCESS)
fprintf(stderr, "发布失败: %s\n", mosquitto_strerror(rc));
else
printf("已上报: %s\n", payload);
}
mosquitto_loop_stop(mosq, true);
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
return 0;
}6. 边缘计算：Z-score 异常检测
在网关上做简单的统计异常检测，减少无效数据上云（降低云端流量和存储成本）。
/*
* zscore_detector.c — Z-score 异常检测（C 实现，无浮点库以外的依赖）
*
* 原理：
* z = (x - μ) / σ
* |z| > threshold → 异常
*
* 滑动窗口实现：维护最近 N 个样本的均值和标准差
*/
#include <stdio.h>
#include <math.h>
#include <string.h>
#define WINDOW_SIZE 100 /* 滑动窗口大小 */
#define Z_THRESHOLD 3.0 /* |z| > 3 认为异常（3σ 原则） */
typedef struct {
double buf[WINDOW_SIZE];
int pos; /* 写入位置（循环） */
int count; /* 已填充的样本数 */
double sum; /* 窗口内总和 */
double sum_sq; /* 窗口内平方和 */
} zscore_ctx_t;
void zscore_init(zscore_ctx_t *ctx) {
memset(ctx, 0, sizeof(*ctx));
}
/* 更新窗口并返回当前样本的 z-score */
double zscore_update(zscore_ctx_t *ctx, double x) {
/* 移除最旧的样本（如果窗口满了） */
if (ctx->count == WINDOW_SIZE) {
double old = ctx->buf[ctx->pos];
ctx->sum -= old;
ctx->sum_sq -= old * old;
} else {
ctx->count++;
}
/* 加入新样本 */
ctx->buf[ctx->pos] = x;
ctx->pos = (ctx->pos + 1) % WINDOW_SIZE;
ctx->sum += x;
ctx->sum_sq += x * x;
/* 窗口样本不足时不判断异常 */
if (ctx->count < 10) return 0.0;
double mean = ctx->sum / ctx->count;
double var = ctx->sum_sq / ctx->count - mean * mean;
double stddev = (var > 0) ? sqrt(var) : 1e-9; /* 防除零 */
return (x - mean) / stddev;
}
int zscore_is_anomaly(zscore_ctx_t *ctx, double x) {
double z = zscore_update(ctx, x);
return fabs(z) > Z_THRESHOLD;
}
/* 示例使用 */
int main(void) {
zscore_ctx_t ctx;
zscore_init(&ctx);
double samples[] = {42.1, 41.9, 42.3, 42.0, 41.8,
42.2, 42.1, 41.9, 42.0, 42.1,
/* 模拟异常点 */
99.5,
/* 恢复正常 */
42.1, 42.0};
for (int i = 0; i < (int)(sizeof(samples)/sizeof(samples[0])); i++) {
double z = zscore_update(&ctx, samples[i]);
printf("样本 %2d: %.2f z=%.3f %s\n",
i, samples[i], z,
fabs(z) > Z_THRESHOLD ? "⚠️ 异常！" : "正常");
}
return 0;
}实测性能（Pynq-Z2，ARM Cortex-A9 @ 667 MHz）：
| 操作 | 耗时 |
|---|---|
zscore_update()（单次） | ~0.08 μs |
| 处理 1ms EtherCAT 数据（8 通道） | ~0.7 μs |
| 1ms 内能处理的最大通道数 | ~11000 |
Z-score 计算开销远小于 EtherCAT 通信和 MQTT 上报，完全在嵌入式端完成没有问题。
7. 数据流全链路：1ms 采样 → 500ms 上报
EtherCAT 从站
↓ 1ms（CPU1 FreeRTOS 实时任务）
环形缓冲区（DDR，OpenAMP 共享内存）
↓ RPMsg（CPU1 → CPU0，每 50 个样本一批）
CPU0 Linux 接收线程
├── Z-score 异常检测（每样本 ~0.08μs）
│ ├── 正常：累积到聚合缓冲区
│ └── 异常：立即推送 MQTT（独立高优先级 topic）
└── 每 500ms：计算 500 个样本的均值/方差
↓
MQTT 上报（QoS 2，JSON，~500ms 周期）
↓ TCP/IP（GEM1）→ LTE 4G → AWS IoT Core实测延迟分解：
| 阶段 | 延迟 | 备注 |
|---|---|---|
| EtherCAT 采样 → 环形缓冲区 | <1ms | CPU1 FreeRTOS，确定性 |
| RPMsg CPU1 → CPU0 | ~0.1ms | OpenAMP IPC |
| Z-score 计算（8 通道） | ~0.01ms | CPU0 |
| MQTT 聚合上报 | 500ms | 设计值（节省流量） |
| MQTT 网络传输（LTE 4G） | ~50ms | 实测（中国/日本区域） |
| 异常立即推送延迟 | ~51ms | 异常点检测到上云 |
8. 本篇 Checklist
- RS-485 总线两端各有 120Ω 终端电阻，
modbus_read_registers()成功返回 -
modbus_new_rtu()连接成功，CRC 错误率 <0.1%（modbus_set_debug()查看） - GEM0 无 IP 地址（
ip addr show eth0只有BROADCAST,MULTICAST无inet） - SOEM
ec_config_init()能发现从站，从站进入 OP 状态 - EtherCAT 循环时间 ≤2ms（99th percentile），用
cyclictest验证系统抖动 - PL IEEE 1588 IP：时间戳锁存寄存器在帧到达后可读
- mosquitto TLS 连接 AWS IoT Core 8883 端口成功
- JSON 上报格式正确，AWS IoT Core 控制台可看到消息
- Z-score 异常检测：插入一个 99.5 的明显异常值，触发告警
9. 下一篇预告
下一篇 《Zynq 实战 29｜系列收官：30 章知识体系回顾 + 常见问题排查 + 进阶路线》 是本系列的最终篇，会做完整的系列总结：
- 29 篇全体系知识地图（一张大表）
- 5 大常见问题排查指南（Bitstream 下载失败 / AXI 超时 / PetaLinux 启动卡住 / DMA 乱序 / 时序违例）
- PS 和 PL 性能优化技巧精要
- Zynq-7000 → UltraScale+ MPSoC → Versal ACAP 进阶路线
参考资料
| 文档号 / 来源 | 名称 | 用途 |
|---|---|---|
| UG585 | Zynq-7000 SoC TRM | GEM（网络控制器）配置，MIO 引脚复用 |
| PG155 | AXI Ethernet Subsystem Product Guide | GEM 连接 PHY，独立于 TCP/IP 协议栈配置 |
| SOEM GitHub | OpenEtherCATsociety/SOEM | EtherCAT 主站开源实现，PetaLinux 移植参考 |
| libmodbus Docs | libmodbus.org | Modbus RTU/TCP API 参考 |
| IEEE 1588-2008 | PTP 精确时间协议标准 | 硬件时间戳实现规范 |
| AWS IoT Developer Guide | docs.aws.amazon.com/iot | MQTT TLS 连接、证书配置 |
这是《Zynq FPGA 嵌入式系统设计实战》系列第 28 篇。 EtherCAT 独占网口和 MQTT TLS 是本篇最常见的两个坑，遇到其他问题欢迎留言。