流处理消息队列后端【免费下载链接】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 Command Line Interface简称 Fluvio CLI是开发者与 Fluvio 集群交互的主要通信机制覆盖数据生产、消费、Topic/Partition 管理以及集群 Profile 配置等全链路操作。本文以 Fluvio CLI 的官方说明为骨架结合仓库源码逐层拆解fluvio produce的命令参数、交互式会话与文件记录摄取机制并补充 Profile 多集群配置与顶层命令结构帮助读者从零掌握这套事件流处理 CLI 的完整用法。Fluvio CLI 与集群的两种核心组件Fluvio CLI 是 Fluvio 集群的主入口几乎所有面向集群的操作都通过它发起。一个 Fluvio 集群由两个核心组件构成分别通过以下命令启动Streaming ControllerSC运行fluvio run sc负责集群的元数据管理与协调Streaming Processing UnitSPU运行fluvio run spu负责实际的数据存储与处理。从仓库源码看这两个启动命令由 fluvio-run crate 提供其中RunCmd枚举定义了spu、sc、metadata、version四个子命令分别转发到fluvio_spu::main_loop与fluvio_sc::start::main_loop。CLI 本身的顶层命令结构定义在 crates/fluvio-cli/src/lib.rs 中通过 clap 派生包含以下几类命令用途consume/produce从 Topic/Partition 读取、写入消息topic/partition管理与查看 Topic 与 Partitionsmartmodule创建与管理 WASM 编译的 SmartModulestable-format别名tf定义消费输出的表格化渲染格式consumer管理与查看消费者remote/home与镜像集群、Home 集群交互profile管理描述已连接集群的 Profilecluster安装或卸载 Fluvio 集群k8s featurebenchmark别名bench运行 Fluvio 基准测试completions为 bash/zsh/fish 生成命令行补全version打印版本信息入口程序位于 crates/fluvio-cli/src/bin/main.rs采用 mimalloc 作为全局分配器未携带参数或传入-h/--help/help时会打印帮助信息后退出。Profiles一套配置管理多个集群Fluvio CLI 使用Profile 文件存储常用连接参数管理员可以通过不同 Profile 方便地在 SC、SPU 或 Kafka 环境之间切换通信目标而无需每次敲入完整地址。Profile 子命令在 crates/fluvio-cli/src/profile/mod.rs 中定义包含子命令说明profile current打印当前上下文名称profile switch name切换到指定 Profileprofile list显示完整 Fluvio 配置PROFILE/CLUSTER/ADDRESS/TLS/INSTALLATIONprofile add name address [type]手动添加 Profile高级用法profile rename重命名 Profileprofile delete/profile delete-cluster删除 Profile 或集群profile sync k8/profile sync local从 Kubernetes 或本地集群同步 Profileprofile export导出 Profile 供其他应用使用默认 TOML 格式源码实现上Profile 与集群的关联关系存储在ConfigFile由fluvio::config提供中Profile 记录引用某个 clustercluster 记录 endpoint 与 TLS 策略。profile list的输出表头为[, PROFILE, CLUSTER, ADDRESS, TLS, INSTALLATION]其中*标记当前激活的 Profile见 crates/fluvio-cli/src/profile/list.rs。profile export会将所选 Profile 关联的 cluster 配置序列化为 TOML 输出便于跨机器迁移配置见 crates/fluvio-cli/src/profile/export.rs。Produce CLI向 Topic/Partition 写入日志记录fluvio produce是数据生产的核心命令可以在单个会话中摄取一个或多个日志记录。其中Topic 与 Partition 是必填参数其余均为可选。Write log records to a topic/partition USAGE: fluvio produce [OPTIONS] --partition integer --topic string FLAGS: -h, --help Prints help information OPTIONS: -t, --topic string Topic name -p, --partition integer Partition id -l, --record-per-line filename Each line is a Record -r, --record-file filename... Entire file is a Record (multiple) -c, --sc host:port Address of Streaming Controller -s, --spu host:port Address of Streaming Processing Unit -k, --kf host:port Address of Kafka Controller -e, --profile profile Profile name消息路由的工作流程produce命令的消息路由遵循SC 定位、SPU 直连的机制查询阶段Produce 命令将 Topic/Partition 请求发送到 SC由 SC 查找持有 leader 分区的 SPU所在位置写入阶段日志记录随后直接发送到该 SPU不再经过 SC 中转Leader 校验记录只有在目标 SPU 恰好是该 Topic/Partition 的 leader 时才会被接受否则被拒绝。向 Kafka 生产消息Produce 命令同样支持向 Kafka 发送日志消息只需指定任意一个 Broker 地址-k/--kf host:port系统会自动识别持有 leader 的 Broker并将日志记录转发到对应位置。从源码看ProduceOpt的参数定义位于 crates/fluvio-cli/src/client/produce/mod.rs其核心执行逻辑会构建TopicProducerConfigBuilder并通过fluvio.topic_producer_with_config(self.topic, config)创建生产者。-c/-s/-k/-e这类目标地址参数在更上层的ClusterTarget中解析最终通过Fluvio::connect_with_config建立连接见 crates/fluvio-cli/src/client/mod.rs。交互式 CLI 会话当produce命令携带 Topic/Partition 启动时会进入交互式会话每一行输入都被解释为一条独立记录逐行发送逐行得到Ok!确认。 fluvio produce -t topic-1 -p 0 line 1 Ok! line 2 Ok! Ctrl-C or Ctrl-D to exit (interactive session)源码中交互模式的判定逻辑在produce_lines→producer_stdin中实现见 crates/fluvio-cli/src/client/produce/mod.rs当启用交互模式时程序会在每次读取前打印提示符发送成功后打印Ok!用户可按Ctrl-C或Ctrl-D退出会话。非交互模式下例如 stdin 来自管道则逐行静默处理输入。从文件摄取记录File Recordsfluvio produce支持两种文件摄取方式Record per File整个文件作为一条记录Record per Line文件的每一行作为一条记录。记录封装Record encapsulation至关重要对于二进制或 JSON 对象对象必须被完整地存入一条记录消费者才能正确解释其内容若按行切分可能破坏对象的完整性。Record per File使用-r/--record-file参数将整个文件封装为单条记录一次命令可以携带一个或多个文件参数可重复。 fluvio produce -t topic-1 -p 0 -r json-file1.txt -r json-file2.txt {name: john doe} Ok! {name: jane doe} Ok!Record per Line使用-l/--record-per-line参数按行摄取记录直到文件结束。注意请使用文本格式文件以便程序正确解释行尾。 fluvio produce -t topic-1 -p 0 -l my-file.txt Lorem Ipsum is simply dummy text Ok! Lorem Ipsum has been the industrys standard since 1500s. Ok! It has survived over five centuries Ok!源码实现上producer-file-io特性默认启用提供了完整的文件输入支持process_raw_file读取整个文件或 stdin作为单条记录发送对应--rawproduce_lines则用BufReader::lines()逐行读取文件并逐条发送并在非AtMostOnce语义下通过join_all等待所有记录确认送达见 crates/fluvio-cli/src/client/produce/mod.rs。源码级进阶Produce 的扩展参数当前仓库中的ProduceOpt相比基础文档提供了更丰富的参数集均定义于 crates/fluvio-cli/src/client/produce/mod.rs参数说明--key key以指定字符串作为 key 发送键值记录--key-separator sep按首个分隔符切分输入为 key/valuekey 决定目标分区--raw将全部输入作为一条记录发送适合二进制文件--compression algo压缩算法none、gzip、snappy、zstd、lz4--linger time发送前等待时间如150ms、20s--batch-size n批量发送的最大记录数--max-request-size bytes累积发送的最大字节数--isolation levelread_committed等待记录提交或read_uncommitted仅等待 leader 接受--delivery-semantic mode默认at-least-onceat-most-once时不等待响应--smartmodule / --smartmodule-path挂载 SmartModule 处理生产数据别名--sm--aggregate-initial value聚合类 SmartModule 的初始累加器值-e/--params keyvalue传给 SmartModule 的额外参数--transforms / --transforms-line数据转换规格YAML 文件或 JSON 字符串-m/--mirror cluster向远程镜像集群生产与-p互斥这些参数通过TopicProducerConfigBuilder逐项注入生产者配置并支持 SmartModule 调用链smartmodule_invocations。此外配合 Topic 管理命令fluvio topic create/delete/describe/list/add-partition/add-mirror见 crates/fluvio-cli/src/client/topic/mod.rs可先创建 Topic支持指定分区数、副本数、保留时间、压缩类型、去重等见 crates/fluvio-cli/src/client/topic/create.rs再通过produce持续写入。配套消费命令与 CLI 生态与produce相对的fluvio consume提供了流式消费能力定义于 crates/fluvio-cli/src/client/consume/mod.rs默认保持活动并持续打印新到达的消息可通过-d在消费完当前全部消息后退出支持-B/--beginning、-H/--head、-T/--tail、--start、--end等多种起始偏移控制以及-O/--output选择dynamic、text、binary、json、raw、table、full_table输出格式-F/--format支持 Handlebars 模板变量{{key}}、{{value}}、{{offset}}、{{partition}}、{{time}}。此外Fluvio CLI 支持外部子命令插件机制当命令名无法匹配内置子命令时会按可执行文件同目录 → 系统 PATH →~/.fluvio/extensions/的顺序查找fluvio-cmd插件并执行见 crates/fluvio-cli/src/lib.rsfluvio completions bash/zsh/fish则可生成 shell 补全脚本例如$ fluvio completions bash ~/fluvio_completions.sh $ echo source ~/fluvio_completions.sh ~/.bashrc至此从集群组件启动、Profile 多集群配置到produce的交互式生产、文件批量摄取与进阶参数Fluvio CLI 的完整使用链路已经清晰可见。读者可依据上述命令与源码路径直接在当前仓库crates/fluvio-cli中查阅实现细节并动手实践。赞分享流处理消息队列后端【免费下载链接】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点击查看免费下载相关推荐CubeFS 集群管理实战cfs-cli cluster 命令完全指南CubeFS 集群管理实战cfs cli cluster 命令完全指南 本篇指南以 CubeFS 官方中文文档 docs zh/source/user gui存储分布式文件系统对象存储云原生GoogleAssistantProxy 消息配置实战用 messagesObj 为 Google Assistant 会话编排欢迎语、追问与告别GoogleAssistantProxy 消息配置实战用 messagesObj 为 Google Assistant 会话编排欢迎语、追问与告别 Googl流处理消息队列后端VoltAgent CLI 评估命令完全指南数据集管理与实验执行实战VoltAgent CLI 评估命令完全指南数据集管理与实验执行实战 导读 本文以 VoltAgent CLI 的 eval 命令族为主线系统讲解从安装初人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆Agent 工作流AI 评测MCP 服务MCP Clients语音上一篇为什么用Zig写CLI工具codex-auth单二进制架构与核心代码实现深度剖析下一篇Qwen-Image-2.1-viggle-turbo ComfyUI教程2个自定义节点2套现成工作流10分钟搭好少步生图创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考