使用 cargo-generate 模板快速开发 Fluvio SmartModule：从模板生成、Rust 编码到 WASM 部署完整指南

流处理消息队列后端【免费下载链接】fluvio event stream processing for developers to collect and transform data in motion to power responsive data intensive applications.项目地址https://gitcode.com/gh_mirrors/fl/fluvio点击查看免费下载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简称 SPUSPU 在把每条记录发送给消费者之前先运行你的 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 中声明的占位符向你逐个提问。理解这些选项是生成出符合预期的工程的第一步占位符类型提示语可选项 / 默认值fluvio-smartmodule-cargo-dependencystringCargo.toml 中fluvio-smartmodule依赖的值自由输入可填 crates.io 版本号如0.8或本地路径依赖project-groupstring请设置一个组名自由输入最终会转成 kebab-case 写入 SmartModule 元数据smartmodule-typestring你需要哪种类型的 SmartModulefilter、map、filter-map、array-map、aggregate默认filtersmartmodule-paramsbool你的 SmartModule 会使用 init 参数吗默认falsesmartmodule-publicbool你的 SmartModule 是公开的吗默认false其中smartmodule-type的选择直接决定生成的src/lib.rs骨架模板用 liquid 模板语法根据该值渲染出对应类型的处理函数同时smartmodule-params决定是否附加一段#[smartmodule(init)]初始化逻辑详见下文。模板生成的项目结构剖析生成后的工程包含以下关键文件各自承担不同职责Cargo.tomlcdylib 与 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要点有三crate-type [cdylib]这是所有 SmartModule 的硬性要求。fluvio-smartmodule 的 README 明确指出 All SmartModules require addingcrate-type [cdylib]to the Cargo.tomlcdylib 才会输出可供 WASM 运行时加载的导出符号依赖注入fluvio-smartmodule是核心运行时 crateserde/serde_json用于 JSON 类 SmartModule如 array-map 解析 JSON 数组当选择了filter类型且需要 init 参数时会自动追加once_cell依赖用于在 WASM 的静态区缓存初始化参数release-ltoprofilelto 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) - Resultbool { 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(OptionRecordData, 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) - ResultOption(OptionRecordData, 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) - ResultVec(OptionRecordData, RecordData) { // Deserialize a JSON array with any kind of values inside let array serde_json::from_slice::Vecserde_json::Value(record.value.as_ref())?; // Convert each JSON value from the array back into a JSON string let strings: VecString 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(OptionRecordData, 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-maptarget/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) - ResultRecordData { // 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: OnceLockString OnceLock::new(); #[smartmodule(init)] fn init(params: SmartModuleExtraParams) - Result() { // You can refer to the example SmartModules in Fluvios 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 中定义其内部是一个BTreeMapString, String提供get(key)、insert(key, value)等访问方法并在min_version 20起扩展了可选的lookback字段用于 Lookback 回溯场景。init 函数执行时机在服务端启动 WASM 模块之后、逐条处理记录之前因此天然适合注入过滤条件、初始累加值等一次性配置。仓库中的完整实践可对照两个示例smartmodule/examples/filter_with_param/src/lib.rs通过params.get(key)设置过滤关键字缺省时回退到asmartmodule/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 中自定义的优化 profilefluvio consume-B表示从头beginning消费--{{smartmodule-type}}是模板占位符——若生成时选择了filter实际命令即为--filtertarget/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 keyvalue以keyvalue形式向 SmartModule 传 init 参数例如fluvio consume topic --smartmodule my_filter -e foobar -e keyvalue--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_smartmodulePredefined名称引用-e参数则汇总为BTreeMapString, String后灌入SmartModuleInvocation.params最终随请求上传 SPU。底层原理#[smartmodule(...)]宏到底生成了什么理解宏展开有助于排查编译与运行问题。smartmodule属性宏定义在 crates/fluvio-smartmodule-derive/src/lib.rs它解析你的函数签名依据类型分发到 generator 目录 下的各生成器类型生成器导出的 WASM 符号no_manglefilterfilter.rsfiltermapmap.rsmapfilter_mapfilter_map.rsfilter_maparray_maparray_map.rsarray_mapaggregateaggregate.rsaggregateinitinit.rsinit其中 transform 类filter/map/filter_map/array_map共享 transform.rs 生成的统一样板其执行流程为解码从 WASM 线性内存读取(ptr, len, version)构造SmartModuleInput再按协议版本解码出VecSmartModuleRecord处理遍历记录调用你的用户函数super::my_filter(record)将返回值写入SmartModuleOutput.successes出错时构造SmartModuleTransformRuntimeError写入output.error并中断编码把SmartModuleOutput重新编码进新的内存块通过宿主导出的copy_records(ptr, len)回调交回运行时函数返回successes.len()。三种非 transform 类型各有特化逻辑filterfilter.rs只把返回Ok(true)的记录推入successesmapmap.rs把返回的(maybe_key, value)写回record.key与record.value后入队array_maparray_map.rs遍历返回的Vec将每个(output_key, output_value)构造为新Record入队aggregateaggregate.rs维护accumulator每轮用SmartModuleAggregateInput.accumulator作累加输入把用户函数返回值写回累加器并同步进输出记录initinit.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 三要素LTASmartModuleExtraParams中min_version 20起的lookback字段即为该能力的载体。宏在解码时按传入的version参数做版本兼容解码确保新旧引擎互通。进阶用 SMDK 构建与发布除直接cargo build外仓库示例工作区还提供了基于 SMDKSmartModule 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 --releasemake 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-unknownCargo.toml 缺少crate-type [cdylib]SmartModule 无法输出可加载的 WASM 符号必须显式声明产物体积过大确认使用--release且命中release-ltoprofilelto true、strip symbols消费命令报参数错误核对当前 Fluvio CLI 版本——新版统一使用--smartmodule-path本地文件或--smartmoduleHub 名称init 参数用-e keyvalue传递aggregate 初始值用--aggregate-initial想给 filter 传参数生成时勾选smartmodule-params true模板会生成#[smartmodule(init)]骨架配合-e keyvalue即可注入运行期配置。至此从模板生成、五种类型编写、WASM 编译到 CLI 接入、再到宏展开原理一条完整的 Fluvio SmartModule 开发链路已经打通。后续可深入仓库的 smartmodule/examples 与 crates/fluvio-smartmodule 源码继续探索 look_back、时间戳感知等更多高级能力。赞分享流处理消息队列后端【免费下载链接】fluvio event stream processing for developers to collect and transform data in motion to power responsive data intensive applications.项目地址https://gitcode.com/gh_mirrors/fl/fluvio点击查看免费下载相关推荐llama-server 从本地部署到 OpenAI 兼容 API 对接llama server 从本地部署到 OpenAI 兼容 API 对接 llama.cpp 的 llama server 是一个纯 C/C 的本地 LLM人工智能大模型模型推理服务推理引擎本地部署后端SyncTeX双向定位JupyterLab LaTeX中实现TeX与PDF同步编辑SyncTeX双向定位JupyterLab LaTeX中实现TeX与PDF同步编辑 JupyterLab LaTeX扩展是一款专为科研工作者和学生打造的高效工vue-lottie常见问题解答解决90%开发者遇到的难题vue lottie常见问题解答解决90%开发者遇到的难题 vue lottie是一款基于Bodymovin的Vue动画渲染组件能够帮助开发者轻松在Vue项上一篇Playnite终极指南如何统一管理你所有游戏平台的完美解决方案下一篇Playnite游戏库管理器一站式免费开源游戏启动中心终极指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考