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

Fluvio Command Line Interface（简称 Fluvio CLI）是开发者与 Fluvio 集群交互的主要通信机制，覆盖数据生产、消费、Topic/Partition 管理以及集群 Profile 配置等全链路操作。本文以 Fluvio CLI 的官方说明为骨架，结合仓库源码逐层拆解fluvio produce的命令参数、交互式会话与文件记录摄取机制，并补充 Profile 多集群配置与顶层命令结构，帮助读者从零掌握这套事件流处理 CLI 的完整用法。

Fluvio CLI 与集群的两种核心组件

Fluvio CLI 是 Fluvio 集群的"主入口"，几乎所有面向集群的操作都通过它发起。一个 Fluvio 集群由两个核心组件构成，分别通过以下命令启动：

• Streaming Controller（SC）：运行fluvio run sc，负责集群的元数据管理与协调；
• Streaming Processing Unit（SPU）：运行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 派生，包含以下几类：

入口程序位于 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 与集群的关联关系存储在ConfigFile（由fluvio::config提供）中：Profile 记录引用某个 cluster，cluster 记录 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 直连"的机制：

1. 查询阶段：Produce 命令将 Topic/Partition 请求发送到 SC，由 SC 查找持有 leader 分区的 SPU所在位置；
2. 写入阶段：日志记录随后直接发送到该 SPU，不再经过 SC 中转；
3. 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 Records）

fluvio 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 industry's standard since 1500's. > Ok! It has survived over five centuries > Ok!

源码实现上，producer-file-io特性（默认启用）提供了完整的文件输入支持：process_raw_file读取整个文件（或 stdin）作为单条记录发送（对应--raw），produce_lines则用BufReader::lines()逐行读取文件并逐条发送，并在非AtMostOnce语义下通过join_all等待所有记录确认送达（见 crates/fluvio-cli/src/client/produce/mod.rs）。

源码级进阶：Produce 的扩展参数

当前仓库中的ProduceOpt相比基础文档提供了更丰富的参数集（均定义于 crates/fluvio-cli/src/client/produce/mod.rs）：

这些参数通过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.rs）；fluvio completions bash/zsh/fish则可生成 shell 补全脚本，例如：

$ fluvio completions bash > ~/fluvio_completions.sh $ echo "source ~/fluvio_completions.sh" >> ~/.bashrc

至此，从集群组件启动、Profile 多集群配置，到produce的交互式生产、文件批量摄取与进阶参数，Fluvio CLI 的完整使用链路已经清晰可见。读者可依据上述命令与源码路径，直接在当前仓库（crates/fluvio-cli）中查阅实现细节并动手实践。

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