企业官网建设流程全解析

• 流处理
• 消息队列
• 后端

Fluvio SmartModule 是运行在 Fluvio 流处理管道中的自定义 WebAssembly 插件，允许开发者在消费数据时对每条记录进行过滤、映射与聚合等实时变换。本指南以仓库内smartmodule/cargo_template/官方模板为核心，讲解如何用cargo-generate一键生成 SmartModule 工程、编写五种类型的处理函数、编译为.wasm并接入 Fluvio CLI 消费端，同时结合仓库源码剖析#[smartmodule(...)]宏的底层展开机制，帮助读者从"会写"进阶到"理解原理"。

SmartModule 是什么：在 SPU 上逐条处理数据的 WASM 插件

根据模板 README.md 的定义：Fluvio SmartModule 是可以用来操纵 Topic 中流式数据的自定义插件。它们用 Rust 编写，编译为 WebAssembly。使用时，你只需把.wasm文件提供给 Fluvio 消费端（consumer），消费端会将其上传到流处理单元（Streaming Processing Unit，简称 SPU），SPU 在把每条记录发送给消费者之前，先运行你的 SmartModule 代码对记录做处理。

这一"数据在途中（data in motion）"被就地变换的模型，是 Fluvio 与普通消息队列的关键区别：处理逻辑被推到服务端、贴近数据源执行，而不是拉回客户端。从仓库源码可以印证这一执行模型：

• 消费端通过 smartmodule_invocation.rs 中的create_smartmodule_from_path读取.wasm文件字节，并使用 gzip 压缩后封装为SmartModuleInvocationWasm::AdHoc，随消费请求一并提交给服务端；
• 服务端 SPU 侧的 smartengine 引擎负责实例化 WASM 模块并对每条记录调用导出的处理函数。

SmartModule 的"入口函数"由过程宏#[smartmodule(kind)]生成：你只需声明处理逻辑，宏会自动补齐 WASM 导出、输入解码、输出编码、错误处理等样板代码（详见后文"底层原理"章节）。

环境准备：安装 cargo-generate 与 WASM 编译目标

模板 README 给出的起步命令非常精简，共两步：

$ cargo install cargo-generate $ cargo generate --git https://github.com/infinyon/fluvio-smartmodule-template

说明：当前仓库内即包含该模板的完整拷贝，位于 smartmodule/cargo_template。如果你使用的是本仓库，可将--git指向本地路径，例如cargo generate --git ./smartmodule/cargo_template，效果一致。

编译目标（WASM target）是容易踩坑的第一步。模板 README 明确提示：要编译 SmartModule，必须先安装wasm32-wasip1目标：

$ rustup target add wasm32-wasip1

如果使用--nowasi相关标志，则改用wasm32-unknown-unknown：

$ rustup target add wasm32-unknown-unknown

模板自带的 rust-toolchain.toml 已经声明了双目标：

[toolchain] channel = "stable" targets = ["wasm32-unknown-unknown", "wasm32-wasip1"]

因此在模板生成的工程中，只要执行rustup target add wasm32-wasip1（或按需添加另一个目标），即可开始编译。仓库的示例工程在 examples/Makefile 中同样通过rustup target add wasm32-wasip1完成目标安装，可以作为参考。

交互式生成：cargo-generate 模板占位符（Placeholder）详解

运行cargo generate后，模板会通过 cargo-generate.toml 中声明的占位符向你逐个提问。理解这些选项，是生成出符合预期的工程的第一步：

其中smartmodule-type的选择直接决定生成的src/lib.rs骨架：模板用 liquid 模板语法根据该值渲染出对应类型的处理函数，同时smartmodule-params决定是否附加一段#[smartmodule(init)]初始化逻辑（详见下文）。

模板生成的项目结构剖析

生成后的工程包含以下关键文件，各自承担不同职责：

Cargo.toml：cdylib 与 release-lto 是硬性要求

模板的 Cargo.toml.liquid 展示了 SmartModule 工程的 Cargo 配置要点：

[package] name = "{{project-name}}" version = "0.1.0" authors = ["{{authors}}"] edition = "2024" [lib] crate-type = ['cdylib'] [dependencies] fluvio-smartmodule = {{fluvio-smartmodule-cargo-dependency}} serde = { version = "1", features = ["derive"] } serde_json = "1" {% if smartmodule-params %}{% if smartmodule-type == "filter" %}once_cell = "1.13.0"{% endif %}{% endif %} [profile.release-lto] inherits = "release" lto = true strip = "symbols"

要点有三：

1. crate-type = ['cdylib']：这是所有 SmartModule 的硬性要求。fluvio-smartmodule 的 README 明确指出 "All SmartModules require addingcrate-type = ['cdylib']to the Cargo.toml"，cdylib 才会输出可供 WASM 运行时加载的导出符号；
2. 依赖注入：fluvio-smartmodule是核心运行时 crate；serde/serde_json用于 JSON 类 SmartModule（如 array-map 解析 JSON 数组）；当选择了filter类型且需要 init 参数时，会自动追加once_cell依赖，用于在 WASM 的静态区缓存初始化参数；
3. release-ltoprofile：lto = true开启链接期优化、strip = "symbols"去除符号表，两者都是为显著缩小 WASM 产物体积而设。仓库示例工作区 smartmodule/examples/Cargo.toml 的注释也写明："lto is need to reduce wasm binary size"，并在[profile.release-lto]中采用完全相同的配置。

SmartModule.toml：面向 SmartModule Hub 的包元数据

模板的 SmartModule.toml 描述 SmartModule 的发布元数据（供 Hub 与 SMDK 工具链识别）：

[package] name = "{{project-name}}" group = "{{project-group | kebab_case}}" version = "0.1.0" apiVersion = "0.1.0" description = "{{project-description}}" license = "Apache-2.0" visibility = {% if smartmodule-public %}"public"{% else %}"private"{% endif %} [[params]] name = "input" description = "input description"

其中visibility由smartmodule-public占位符控制：默认private（仅自己可见），选是则为public（可发布到 Hub 供他人复用）。[[params]]段声明 SmartModule 对外暴露的参数入口。

rust-toolchain.toml：锁定 stable 与双 WASM 目标

如前述，rust-toolchain.toml 声明channel = "stable"与wasm32-unknown-unknown、wasm32-wasip1两个目标，确保团队协作时编译环境一致。

五种 SmartModule 类型：模板源码与仓库示例对照

smartmodule-type占位符的选择决定生成的 src/lib.rs 骨架。下面逐一解读五种类型（该文件是 liquid 模板，{% if %}分支即用户选择后实际渲染出的代码）。

Filter：按条件决定记录去留

过滤器接收每条记录的引用，返回true（保留）或false（丢弃）。模板生成的 filter 骨架：

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord}; #[smartmodule(filter)] pub fn filter(record: &SmartModuleRecord) -> Result<bool> { let string = std::str::from_utf8(record.value.as_ref())?; Ok(string.contains('a')) }

即只保留值中含有字母a的记录。模板 README 给出的等价示例用SimpleRecord与String::from_utf8_lossy，保留了值中含z的记录：

use fluvio_smartmodule::{smartmodule, SimpleRecord}; #[smartmodule(filter)] pub fn my_filter(record: &SimpleRecord) -> bool { let value = String::from_utf8_lossy(record.value.as_ref()); value.contains('z') }

仓库的真实示例 smartmodule/examples/filter/src/lib.rs 与模板骨架完全同构，可直接对照学习。

Map：一对一变换记录

Map 对每条记录做 1:1 变换，可改写 key 与 value。模板生成的 map 骨架：

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord, RecordData}; #[smartmodule(map)] pub fn map(record: &SmartModuleRecord) -> Result<(Option<RecordData>, RecordData)> { let key = record.key.clone(); let string = std::str::from_utf8(record.value.as_ref())?; let int = string.parse::<i32>()?; let value = (int * 2).to_string(); Ok((key, value.into())) }

该示例把每条记录的值解析为整数后乘以 2，并保留原 key。仓库示例 smartmodule/examples/map/src/lib.rs 则展示了把 value 全部转为大写字母的等价实现（value.make_ascii_uppercase()），返回(key, value.into())。

FilterMap：过滤与映射二合一

FilterMap 兼具两者能力：返回Some(...)时输出变换后的记录，返回None时丢弃该条记录，从而实现"只处理偶数并除以 2"这类条件变换。模板骨架：

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, RecordData, Result}; #[smartmodule(filter_map)] pub fn filter_map(record: &SmartModuleRecord) -> Result<Option<(Option<RecordData>, RecordData)>> { let key = record.key.clone(); let string = String::from_utf8_lossy(record.value.as_ref()).to_string(); let int: i32 = string.parse()?; if int % 2 == 0 { let output = int / 2; Ok(Some((key.clone(), RecordData::from(output.to_string())))) } else { Ok(None) } }

仓库示例 smartmodule/examples/filter_map/src/lib.rs 的注释"过滤掉所有奇数，把所有偶数除以 2"正是对该行为的精确定义。

ArrayMap：一进多出，拆解复合记录

ArrayMap 把一条输入记录展开为零到多条输出记录，适合将逻辑上包含多个数据点的记录拆分。模板骨架演示了最典型的场景——把 JSON 数组拆成多条独立记录：

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord, RecordData}; #[smartmodule(array_map)] pub fn array_map(record: &SmartModuleRecord) -> Result<Vec<(Option<RecordData>, RecordData)>> { // Deserialize a JSON array with any kind of values inside let array = serde_json::from_slice::<Vec<serde_json::Value>>(record.value.as_ref())?; // Convert each JSON value from the array back into a JSON string let strings: Vec<String> = array .into_iter() .map(|value| serde_json::to_string(&value)) .collect::<core::result::Result<_, _>>()?; // Create one record from each JSON string to send let kvs: Vec<(Option<RecordData>, RecordData)> = strings .into_iter() .map(|s| (None, RecordData::from(s))) .collect(); Ok(kvs) }

仓库中 smartmodule/examples/array_map_json_array/src/lib.rs 提供了可运行验证的完整案例：向 Topic 生产["Apple", "Banana", Cranberry]，经 array-map 消费后输出"Apple"、"Banana"、"Cranberry"三条独立记录。该模块文档注释中的完整命令序列（fluvio topic create array-map-array→fluvio produce array-map-array→fluvio consume array-map-array -B --array-map=target/wasm32-wasip1/release/fluvio_wasm_array_map_array.wasm）可作为端到端自测清单。

Aggregate：累加器驱动的流式聚合

Aggregate 维护一个跨记录持续传递的累加器（accumulator）：每次调用接收当前累加值与当前记录，返回新的累加值，输出流即为每一步的累加结果。模板骨架实现求和：

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord, RecordData}; #[smartmodule(aggregate)] pub fn aggregate(accumulator: RecordData, current: &SmartModuleRecord) -> Result<RecordData> { // Parse the accumulator and current record as strings let accumulator_string = std::str::from_utf8(accumulator.as_ref())?; let current_string = std::str::from_utf8(current.value.as_ref())?; // Parse the strings into integers let accumulator_int = accumulator_string.trim().parse::<i32>().unwrap_or(0); let current_int = current_string.trim().parse::<i32>()?; // Take the sum of the two integers and return it as a string let sum = accumulator_int + current_int; Ok(sum.to_string().into()) }

仓库中 smartmodule/examples/aggregate-sum/src/lib.rs 与该骨架完全一致，而 smartmodule/examples/aggregate/src/lib.rs 则演示字符串拼接版聚合（"a"、"b"累加成"1a"、"1ab"），展示了同一接口的多种语义。

init 参数：为 SmartModule 注入运行期配置

当生成时选择smartmodule-params = true，模板会在骨架末尾附加一段初始化函数，用于在正式开始处理前配置状态。filter 类型生成的 init 代码通过OnceLock缓存一个过滤关键字：

static CRITERIA: OnceLock<String> = OnceLock::new(); #[smartmodule(init)] fn init(params: SmartModuleExtraParams) -> Result<()> { // You can refer to the example SmartModules in Fluvio's GitHub Repository if let Some(key) = params.get("key") { CRITERIA.set(key.clone()).map_err(|err| eyre!("failed setting key: {:#?}", err)) } else { Err(SmartModuleInitError::MissingParam("key".to_string()).into()) } }

参数类型SmartModuleExtraParams在 crates/fluvio-smartmodule/src/input.rs 中定义：其内部是一个BTreeMap<String, String>，提供get(key)、insert(key, value)等访问方法，并在min_version = 20起扩展了可选的lookback字段（用于 Lookback 回溯场景）。init 函数执行时机在服务端启动 WASM 模块之后、逐条处理记录之前，因此天然适合注入过滤条件、初始累加值等一次性配置。

仓库中的完整实践可对照两个示例：

• smartmodule/examples/filter_with_param/src/lib.rs：通过params.get("key")设置过滤关键字，缺省时回退到"a"；
• smartmodule/examples/aggregate-init/src/lib.rs：通过params.get("initial_value")设置聚合初始值，并用自定义UseOnce包装保证初始值只消费一次。

编译与接入 Fluvio CLI：把 WASM 跑在消费管道上

模板 README 给出了从创建 Topic 到消费验证的完整闭环：

$ fluvio topic create smartmodule-test $ cargo build --release $ fluvio consume smartmodule-test -B --{{smartmodule-type}}="target/wasm32-wasip1/release-lto/{{project-name}}"

• fluvio topic create smartmodule-test：创建测试 Topic（需先完成 Fluvio 集群与 CLI 的安装与登录）；
• cargo build --release：编译 WASM。产物路径为target/wasm32-wasip1/release-lto/<project-name>，其中release-lto即模板 Cargo.toml 中自定义的优化 profile；
• fluvio consume：-B表示从头（beginning）消费，--{{smartmodule-type}}是模板占位符——若生成时选择了filter，实际命令即为--filter="target/wasm32-wasip1/release-lto/my-filter"，其余类型同理（--map、--filter-map、--array-map、--aggregate）。模板 README 之所以这样书写，是因为该命令面向的是提供类型专用标志的 Fluvio CLI 版本。

当前仓库的 CLI 消费端参数在此基础上做了统一：查看 crates/fluvio-cli/src/client/consume/mod.rs 可见，fluvio consume现已同时支持：

• --smartmodule <name>/ 别名--sm：按名称引用已注册（预定义）的 SmartModule；
• --smartmodule-path <path>/ 别名--sm_path：直接指定本地.wasm文件路径（推荐用于本指南场景）；
• -e, --params <key=value>：以key=value形式向 SmartModule 传 init 参数，例如fluvio consume topic --smartmodule my_filter -e foo=bar -e key=value；
• --aggregate-initial <value>/ 别名--a-init：为 aggregate 类 SmartModule 指定初始累加值（requires = "aggregate_group"，即必须配合 aggregate 类型的 SmartModule 使用）。

消费请求中的 SmartModule 装配逻辑位于 smartmodule_invocation.rs：--smartmodule-path走create_smartmodule_from_path（读文件 + gzip 压缩为AdHoc字节流），--smartmodule走create_smartmodule（Predefined名称引用），-e参数则汇总为BTreeMap<String, String>后灌入SmartModuleInvocation.params，最终随请求上传 SPU。

底层原理：#[smartmodule(...)]宏到底生成了什么

理解宏展开有助于排查编译与运行问题。smartmodule属性宏定义在 crates/fluvio-smartmodule-derive/src/lib.rs，它解析你的函数签名，依据类型分发到 generator 目录 下的各生成器：

其中 transform 类（filter/map/filter_map/array_map）共享 transform.rs 生成的统一样板，其执行流程为：

1. 解码：从 WASM 线性内存读取(ptr, len, version)，构造SmartModuleInput，再按协议版本解码出Vec<SmartModuleRecord>；
2. 处理：遍历记录，调用你的用户函数（super::my_filter(&record)），将返回值写入SmartModuleOutput.successes；出错时构造SmartModuleTransformRuntimeError写入output.error并中断；
3. 编码：把SmartModuleOutput重新编码进新的内存块，通过宿主导出的copy_records(ptr, len)回调交回运行时，函数返回successes.len()。

三种非 transform 类型各有特化逻辑：

• filter（filter.rs）：只把返回Ok(true)的记录推入successes；
• map（map.rs）：把返回的(maybe_key, value)写回record.key与record.value后入队；
• array_map（array_map.rs）：遍历返回的Vec，将每个(output_key, output_value)构造为新Record入队；
• aggregate（aggregate.rs）：维护accumulator，每轮用SmartModuleAggregateInput.accumulator作累加输入，把用户函数返回值写回累加器并同步进输出记录；
• init（init.rs）：解码SmartModuleInitInput后调用你的init(input.params)，成功返回0，失败则编码SmartModuleInitOutput并返回SmartModuleInitErrorStatus::InitError。

此外，crates/fluvio-smartmodule/src/memory.rs 导出了alloc(len)与dealloc(ptr, size)两个no_mangle内存管理函数，供宿主运行时在 WASM 线性内存中分配/释放输入输出缓冲——这是宏样板与宿主之间的"内存契约"。

记录模型：SmartModuleRecord 与协议版本

宏解码出的SmartModuleRecord定义在 crates/fluvio-smartmodule/src/lib.rs，它是对底层Record的包装，额外携带批次级元信息并暴露便捷方法：

• offset()：由批次base_offset与记录内偏移增量累加得出；
• timestamp()：由批次base_timestamp与记录时间戳增量累加得出；
• key()/value()：访问记录的键与值（RecordData类型）。

输入输出结构（SmartModuleInput、SmartModuleOutput、SmartModuleAggregateInput/Output、SmartModuleInitInput/Output）集中在 input.rs 与 output.rs。其中SMARTMODULE_TIMESTAMPS_VERSION = 22（见 input.rs）标志着当前协议版本已支持 Lookback、时间戳与 Age 三要素（LTA），SmartModuleExtraParams中min_version = 20起的lookback字段即为该能力的载体。宏在解码时按传入的version参数做版本兼容解码，确保新旧引擎互通。

进阶：用 SMDK 构建与发布

除直接cargo build外，仓库示例工作区还提供了基于 SMDK（SmartModule Development Kit）的构建路径。smartmodule/examples/Makefile 展示了两种方式：

setup: rustup target add wasm32-wasip1 build: setup $(SUBDIRS) $(SUBDIRS): cd $@ && ${SMDK_BIN} build ${SMDK_OPT} build-raw: setup cargo build --release

• make build：遍历smartmodule/examples/下每个子目录，调用smdk build逐个构建（可利用SMDK_OPT传入额外选项）；
• make build-raw：在 examples 工作区根目录直接cargo build --release。

若要将 SmartModule 发布到 Hub 供其他用户通过--smartmodule <name>引用，则需要完善模板生成的 SmartModule.toml（组名、描述、可见性等），并借助 SMDK 的发布命令完成打包与上传——这正是smartmodule-public占位符存在的意义。

常见问题速查

• 找不到wasm32-wasip1目标：执行rustup target add wasm32-wasip1（或按--nowasi场景改加wasm32-unknown-unknown）；
• Cargo.toml 缺少crate-type = ['cdylib']：SmartModule 无法输出可加载的 WASM 符号，必须显式声明；
• 产物体积过大：确认使用--release且命中release-ltoprofile（lto = true、strip = "symbols"）；
• 消费命令报参数错误：核对当前 Fluvio CLI 版本——新版统一使用--smartmodule-path（本地文件）或--smartmodule（Hub 名称），init 参数用-e key=value传递，aggregate 初始值用--aggregate-initial；
• 想给 filter 传参数：生成时勾选smartmodule-params = true，模板会生成#[smartmodule(init)]骨架，配合-e key=value即可注入运行期配置。

至此，从模板生成、五种类型编写、WASM 编译到 CLI 接入、再到宏展开原理，一条完整的 Fluvio SmartModule 开发链路已经打通。后续可深入仓库的 smartmodule/examples 与 crates/fluvio-smartmodule 源码，继续探索 look_back、时间戳感知等更多高级能力。

• 流处理
• 消息队列
• 后端