mq-bridge-app

mq-bridge

跨越数据流 · 库

一款高速 Rust 引擎，可在代理、数据库、文件和 HTTP 之间传输与重塑数据。可在命令行中试用，也可集成到 Rust、Python 或 Node.js 服务中。

🖥️ mq-bridge-app · 📊 ETL 基准测试 · 🏗️ 架构 · ⚙️ 配置 · 📚 中间件与端点参考

什么是 mq-bridge？

mq-bridge 是一个异步 Rust 库，可在系统之间传输消息和数据：Kafka、NATS、RabbitMQ、MQTT、MongoDB、PostgreSQL/MySQL/SQLite、Postgres CDC、ClickHouse、Redis Streams、HTTP、WebSocket、gRPC、ZeroMQ、AWS SQS/SNS、IBM MQ、云对象存储、文件和内存通道。每个端点都遵循相同的批次形状 receive_batch / send_batch 特征，因此任何源都能向任何接收端输送数据。

路由将一个输入连接到一个输出，并在传输过程中处理数据：通过配置即可转换、筛选、分流、扇出、丰富数据、重试、限速、去重，或响应请求。引擎速度很快：类型化 transform 的运行速度约为每秒 160 万行，在相同任务上比 Redpanda Connect 快约 19.5 倍。switch、fanout 和 sequence 等结构化端点可在路由中自由嵌套；投递组合错误会在路由启动时被发现，而不是等到生产环境。配置无法满足需求时，可在 Rust、Python 或 Node.js 中添加处理器。批处理、ack/nack、提交顺序和代理 I/O 都由基于 Tokio 的核心处理，因此你的代码只需使用普通的 CanonicalMessage，无需处理 Kafka 偏移量或 AMQP nack。

在命令行中试用，集成到你的服务中：mq-bridge-app 可通过命令行（mqb）、服务器、桌面应用或 MCP 服务运行路由；当你需要更低延迟或希望将流水线放入生产代码时，同一份配置也可作为库运行在 Rust、Python 或 Node.js 服务中。无需运维守护进程或平台。

**连接不同生态：**两个插件新增了 100 多种输入和输出：Redpanda Connect 的组件和 faucet 的连接器。它们仅在路由使用时加载，并在同一进程中运行，因此一条路由可以使用一个框架读取、使用第二个框架处理、再使用第三个框架写入，全程共用一条 ack 链，并由 mq-bridge 的重试、DLQ 和去重机制提供保护。请参阅在一条路由中组合不同生态。

使用场景

问题	提供可运行示例的指南
RAG 或语义搜索索引与其 Postgres 源数据不同步	让向量索引与 Postgres 保持同步：CDC → embeddings → Qdrant
搜索结果显示了 MongoDB 中已不存在的内容	让搜索索引与 MongoDB 保持同步：change stream → Meilisearch
数据仓库的数据落后一天，遗漏了期间发生的所有更改	将 Postgres 更改流式传输到数据仓库：CDC → Bloblang → Snowflake
需要根据自由文本消息的内容进行路由	使用大模型对消息分类并路由
需要为事件添加数据库记录和逐键总计，以便据此采取行动	在 Postgres 中丰富事件并添加逐键总计
服务必须以原子方式更新数据库并发布事件	使用 Postgres 和 Kafka 实现事务型 outbox
Kafka、NATS 或 RabbitMQ 消费者的测试需要运行中的代理	无需代理即可测试消息处理程序

快速开始

为你的语言安装库：

语言	软件包	安装
Rust	mq-bridge	cargo add mq-bridge --features kafka,nats,yaml
Python	mq-bridge (PyPI)	pip install mq-bridge
Node.js	mq-bridge (npm)	npm install mq-bridge
C / C++	mq_bridge.h	cargo build --release -p mq-bridge-c（从源代码构建）

在 routes.yaml 中描述一条路由——此处为带重试的 Kafka → NATS：

kafka_to_nats:
  input:
    kafka: { url: "localhost:9092", topic: "orders", group_id: "bridge" }
  output:
    middlewares:
      - retry: { max_attempts: 5 }
    nats: { url: "nats://localhost:4222", subject: "orders.processed" }

运行：

from mq_bridge import Route
Route.from_file("routes.yaml", "kafka_to_nats").run()  # blocks; use start() to keep going

import { Route } from "mq-bridge";
Route.fromFile("routes.yaml", "kafka_to_nats").start();

mq_bridge::deploy_file("routes.yaml").await?; // every route in the file, in the background

需要在中间加入业务逻辑时，可附加一个处理程序（with_handler / withHandler）。完全不想编写代码？ mq-bridge-app 可通过命令行或桌面 UI 运行同一引擎：

mqb copy 'kafka://localhost:9092?topic=orders' 'nats://localhost:4222?subject=orders.processed'

功能一览

每一行功能都可通过 Rust crate、Python 和 Node.js 软件包以及零代码应用使用；只有自行构建 Rust crate 时才需要 Cargo feature。链接的手册同时介绍了库和应用。

能力	方式（端点 / 中间件 · Cargo 功能）	文档
消息代理：Kafka、NATS（JetStream）、RabbitMQ（AMQP）、MQTT、Redis Streams、AWS SQS/SNS、IBM MQ、ZeroMQ	kafka、nats、amqp、mqtt、redis_streams、aws、ibmmq、zeromq 端点 · 功能 kafka、nats、amqp、mqtt、redis-streams、aws、ibm-mq、zeromq	连接器
变更数据捕获（CDC）：将 Postgres 或 MongoDB 的变更流式传输到 Kafka、文件、数据仓库等	postgres_cdc（逻辑复制，pgoutput） · postgres-cdc；mongodb 配合 consume: capture_all / capture_new（变更流） · mongodb	Postgres CDC 教程 · MongoDB
SQL 数据库：PostgreSQL、MySQL、MariaDB、SQLite，可作为源（游标轮询）或接收端（插入 / upsert）	sqlx 端点 · sqlx	PostgreSQL / MySQL / SQLite · Upsert
MongoDB作为源或接收端	mongodb 端点 · mongodb	MongoDB
ClickHouse批量插入和可恢复读取	clickhouse 端点 · clickhouse	ClickHouse
搜索引擎及其他批量文档 API（Meilisearch、Typesense、Elasticsearch、Qdrant、PostgREST / Supabase）	http_bulk 端点 · http-bulk	HTTP 批量处理
通过兼容 OpenAI 的 API 使用 大模型和嵌入（OpenAI、Ollama、vLLM 等）：对每条消息进行分类、摘要或嵌入	openai_chat、openai_embeddings 端点 · http-bulk	兼容 OpenAI 的 API · 大模型操作指南
对象存储：S3（及兼容 S3 的存储，例如 R2）、GCS、Azure Blob 或本地目录，格式为 Parquet、JSONL 或 CSV	object_store 端点，使用 format: parquet | raw | json | csv · object-store、parquet	对象存储
文件：读取和写入 CSV 与 JSONL	file 端点 · 内置	文件
分析 / 数据仓库：Snowflake、BigQuery、Databricks / Spark、Athena / Trino、DuckDB，通过 S3/GCS/Azure 上的 Parquet。这些不是原生连接器：mq-bridge 写入 Parquet 文件，数据仓库负责加载或查询这些文件	object_store，使用 format: parquet · parquet	Snowflake · BigQuery · Databricks / Spark · Athena / Trino · DuckDB
HTTP、gRPC、WebSocket，可作为服务端或客户端，包括请求 / 响应	http、grpc、websocket 端点 · http、grpc、websocket	HTTP · gRPC · 请求 / 响应
架构验证和类型强制转换（JSON Schema 子集、字段映射）	transform 中间件 · 内置	转换与架构映射
可靠性：重试、死信队列、去重、限流	retry、dlq、deduplication、limiter 中间件 · dedup 用于去重	中间件参考 · 投递保证
无需消息代理的测试：使用内存通道运行路由或处理器，无需 Docker	memory 端点 · 内置	嵌入库
无需编写代码的 ETL	mq-bridge-app：mqb copy <from> <to>、YAML 路由、桌面 UI、MCP 服务	快速入门：mqb copy
通过插件接入更多系统：Apache Pulsar、Meilisearch、Redpanda Connect 的输入、输出和处理器，以及 faucet 的源和接收端（MySQL / SQL Server / Oracle CDC、Snowflake、BigQuery、Iceberg、Delta、GraphQL API 等）	原生插件 · plugin	插件端点 · Connect 插件：输入、输出、处理器 · faucet 插件
Connect 插件仅链接不含 Redpanda Community License 代码的 Redpanda Connect 组件（捆绑代码采用 Apache-2.0 和 MIT 许可证），因此不包含部分上游组件，例如 Kafka、AWS 和 Snowflake 组件；请参阅未包含的内容。原生端点和 faucet 插件可弥补其中的大部分空缺。

在同一条路由中组合多个生态系统

没有任何单一框架能涵盖所有连接器和处理器。mq-bridge 在一条路由中组合了三者：

部件来源	提供的功能	数量
mq-bridge，原生	Kafka、NATS、RabbitMQ、MQTT、IBM MQ、Postgres 和 MongoDB CDC、SQL、对象存储、分页 HTTP API（http_bulk）、HTTP / gRPC / WebSocket 服务端，……；transform、switch、fanout、aggregate、lookup、重试、DLQ、去重	20+ 种传输方式
Connect 插件	Redpanda Connect 的组件；可在任意端点上作为中间件使用的 Bloblang 处理器	56 种输入、68 种输出、86 种处理器
faucet 插件	faucet 的连接器：MySQL / SQL Server / Oracle CDC、Snowflake、BigQuery、Iceberg、Delta、GraphQL、Singer tap；写入受支持的接收端时实现恰好一次语义	41 种源、34 种接收端

只有当路由指定插件时，插件才会加载，并在进程内运行。所有部件都支持确认、提交、retry、dlq 和 deduplication，插件端点也可以嵌套在 switch、fanout 和 sequence 中。如果某个系统同时有两种实现，请优先使用原生端点。

场景	原生	Connect	faucet
Postgres 写入 Snowflake，先回填再同步变更，确保恰好一次	postgres_cdc、retry、dlq	connect_mapping	snowflake 接收端
MySQL、SQL Server 或 Oracle 的变更写入 Kafka	kafka 输出、transform	connect_mapping	mysql-cdc、mssql-cdc、oracle-cdc
传感器数据：实时统计、向时序数据库发送告警，并将所有数据写入数据湖	mqtt、aggregate、switch、fanout	questdb 输出	iceberg 或 delta 接收端

使用 brew、conda、pip 或 npm 安装 Connect 插件（安装说明）；使用 pip install mq-bridge-faucet 或 npm install mq-bridge-faucet 安装 faucet 插件（README）。

面向 AI Agent：llms.txt 是文档的精选索引；运行 npx skills add marcomq/mq-bridge 可安装一个 skill， 帮助 Claude Code、Codex 和其他 Agent 使用 mqb。

基准测试

通过零代码 mq-bridge-app 进行同类 ETL 对比；所有数据、基准和命令均见 ETL 基准测试工具：

场景	mq-bridge	对比对象
CSV → JSONL，100 万行混合类型数据（约 116 MiB）	3,134,796 行/秒，约 29 MiB 内存	Meltano（tap-csv → target-jsonl）：9,771 行/秒，约 444 MiB 内存 — mq-bridge 快约 321 倍 DuckDB，使用所有核心：2,109,704 行/秒 — mq-bridge 快约 1.5 倍，内存占用少约 18 倍
Kafka → 文件，100 万行，无转换	878,105 行/秒，约 160 MiB 内存	Sea Streamer，mimalloc 构建版本：492,465 行/秒，约 894 MiB 内存 — mq-bridge 快约 1.8 倍（相较于其默认分配器构建版本也快约 1.8 倍）
CSV → JSONL，100 万行，使用 transform 进行类型转换	1,594,896 行/秒	Redpanda Connect 4.112.0，Bloblang mapping：81,893 行/秒 — mq-bridge 快约 19.5 倍，并验证输出完全相同
Kafka → JSONL，100 万行，投影	766,280 行/秒	Redpanda Connect 4.112.0：65,741 行/秒 — mq-bridge 快约 11.7 倍，内存占用不到一半，写入接收端的字节完全相同

CSV 行复用了一个已公开发布的 100 万行 CSV → JSONL 工作负载，使用逐字节相同的测试文件：7 个混合类型列，随机种子为 42，121,981,421 字节，SHA-256 a84894e0…0c45b221（完整定义）。测量整个进程的实际运行时间，不进行转换；Meltano 在同一台机器上使用同一文件运行。

所有数据均来自 mq-bridge 0.4.20，Redpanda Connect 数据除外；这些数据是在之后的一次测试中与 mq-bridge 原生数据一同测得的（场景 10）。DuckDB 数据表示转换本身的吞吐量上限，而非 ETL 工具。

**外部基准测试：**Rust 库（mq-bridge）和 Python 绑定（mq-bridge-py）均列入第三方 http-arena.com 排行榜，该榜单按每秒请求数比较 HTTP 框架（数据实时更新，因此排名会随时间变化）。它测量 mq-bridge 的 HTTP（和 WebSocket）服务路径，而这只是众多端点之一，并不代表消息代理或 ETL 吞吐量；相关数据见基准测试。本地对比测试工具见 Python 分析说明。

为什么选择 mq-bridge

无论使用哪种语言调用，一个引擎都能提供：

• 20 多种原生传输方式、通过插件支持 100 多种，统一 API：上面列出的每个连接器，以及 dir_spool（具备崩溃安全能力的目录 FIFO 队列），都采用相同的 receive_batch / send_batch 格式。
• 覆盖 Redpanda Connect 生态：Connect 插件提供经过生产环境验证的 Redpanda Connect 组件——56 个输入端点、68 个输出端点，以及 86 个处理器，其中大多数也能作为任意端点的中间件运行；mq-bridge 则负责路由、批处理、重试、DLQ 和去重。
• 覆盖 faucet 生态：faucet 插件将 faucet 的 41 种源和 34 种汇作为一个 faucet 端点提供：MySQL / SQL Server / Oracle CDC、Snowflake、BigQuery、Iceberg、Delta、GraphQL API 和 Singer taps，并支持向支持该功能的汇传递带提交令牌的恰好一次消息。
• 一个路由整合三种框架的优势：原生端点、Connect 和 faucet 组件可在一个路由和一个进程中自由混用；请参阅在一个路由中组合不同生态。
• 变更数据捕获：将 Postgres（逻辑复制 / pgoutput）和 MongoDB（变更流）中的行级变更流式传输为带操作标记的扁平行。
• 重启后安全投递：支持批处理的 ack/nack，以及针对累积确认型 Broker 的提交顺序控制；集成测试套件显示，Broker 在处理过程中重启不会造成数据丢失，其中包括 Postgres CDC 重启安全测试。
• 可靠性中间件，而非框架：重试、死信队列、去重、速率限制和 Cookie/会话持久化可用于包装任意端点。
• 一个引擎支持多种语言：同一 Rust 核心提供原生 Python 和 Node.js 绑定；路由、批处理和 Broker I/O 均由 Rust 处理。
• 统一配置，处处支持 TLS：单个 TlsConfig 配置块（CA 证书包、用于 mTLS 的客户端证书/密钥、不安全跳过验证）可在各种传输方式中复用。
• 可自托管，无需守护进程：在可选 UI 中生成配置，将其粘贴到代码中，然后在进程内运行。无需托管控制平面，也无需单独的调度器。

不想编写代码？ mq-bridge-app 运行完全相同的引擎，是一款独立的零代码 ETL 服务，完全通过 YAML 或环境变量配置。它附带 Postman 风格的 UI，用于针对路由构建、发送和检查消息，还可导入 Postman 集合和 AsyncAPI 文档，以生成路由和端点的初始配置。有关该应用的一切——安装、CLI、UI、操作指南和调优——请参阅📖手册。

何时使用 mq-bridge

• 混合消息传递：连接使用不同协议的系统（例如，将 MQTT 连接到 Kafka），无需为每一对系统编写自定义适配器。
• 批处理密集型管道：以批次传输消息来提高吞吐量，同时保留逐条消息的 ack/nack 决策。
• 基础设施抽象：让业务逻辑基于 CanonicalMessage 编写，之后可替换底层传输方式。
• 弹性管道：在端点周围统一应用重试、DLQ、去重、限流器和 Cookie/会话行为。
• 数据库集成：将数据库与消息 Broker 结合，例如将消息写入 SQL/MongoDB，或将 outbox 行转发到 Broker。
• Sidecar / 网关：在另一项服务旁运行桥接器，在消息到达核心应用前接收、过滤并路由消息。

何时不应使用 mq-bridge

• 有状态流处理：如果需要进行窗口计算、联接或随时间变化的复杂聚合，更适合使用专用的流处理引擎。
• 领域聚合管理：如果需要一个框架来管理领域聚合（事件溯源）的生命周期、版本控制和重放，请使用专用库。mq-bridge 处理的是总线，而不是实体。
• 特定协议的高级功能：mq-bridge 有意只提供一组通用功能：发布/消费、在支持时使用发布/订阅、在支持时使用请求-应答、批处理、中间件以及 ack/nack 处理。如果应用依赖某个代理的特定高级功能，直接使用该代理的原生客户端可能更合适。

文档

**📖 mq-bridge 手册**介绍了该库和应用。连接器、中间件和 调优在两者中使用相同的设置。

我想要……	阅读
嵌入该库、编写处理程序、通过代码发布	快速入门：库 · 嵌入该库 · Bindings API · docs.rs
无需编写代码即可迁移数据	快速入门：mqb copy · 桌面 UI · CLI · MCP 服务
配置连接器	连接器，每个连接器均附有 YAML 和 URL 示例
了解各后端支持的功能（订阅者模式、请求-应答、nack、CDC 与轮询）	概览与功能 · 功能表（确认、提交顺序、应答、TLS）
添加重试、DLQ、去重、转换和路由	操作指南 · 中间件与结构化端点
了解投递保证和幂等接收端	投递保证
构建请求/应答或 CQRS 风格的流程	请求/应答 · 嵌入该库：CQRS
调优吞吐量	性能调优
编写自己的端点、中间件或插件	EXTENDING.md · PLUGINS.md
了解内部实现	ARCHITECTURE.md · CONFIGURATION.md

状态

mq-bridge 创建于 2025 年。其可靠性行为通过自动化集成和性能测试套件进行验证，覆盖所有受支持的端点，以及各端点支持的队列和订阅者模式：

• 所有端点均证实，代理在消息传输过程中重启不会导致数据丢失；针对 MQTT 发布确认进行了强化，直到混沌测试将传输过程中发生的数据丢失降至 零。
• Postgres CDC 设有重启安全测试：数据库重启后，未 ack 且正在传输的批次会重新投递，不会丢失数据或出现间隙。

该库以 Rust 库的形式用于 armature，并以 Python 库的形式用于 omniload。

稳定性和 Rust 版本

自 1.0 起，Rust API 遵循语义化版本规范：crate 根模块以及 models、traits、errors、outcomes、extensions、plugin（SDK）、type_handler、publisher、route 和 checkpoint 模块。实现模块 endpoints、middleware、support 和 test_utils 可能会在次要版本中变更，重新导出的驱动 crate（mq_bridge::sqlx、mq_bridge::rdkafka 等）的主版本号也可能变更。

次要版本可以新增配置字段、配置枚举或错误枚举的变体，也可以提高最低 Rust 版本。使用 ..Default::default() 编写配置结构体，并在这些枚举的 match 中提供 _ 分支的代码，在此类版本升级后仍可不经修改直接编译。

最低 Rust 版本为 1.89。aws、sqlx 和 postgres-cdc 功能以及 full 功能需要 Rust 1.94.1；zeromq-omq 需要 Rust 1.93。

运行测试

项目包含集成测试和性能测试。大多数后端测试都需要 Docker。

要运行所有受支持后端的性能基准测试：

cargo test --test integration_test --release -- --ignored --nocapture --test-threads=1

要运行 criterion 基准测试：

cargo bench --features "full"

Criterion 的测试结果会因运行机器而异；如需可比较的吞吐量数据，请使用上面的集成性能测试，或参阅 ETL 基准测试。

贡献

欢迎贡献。设置说明、代码风格和拉取请求指南请参阅 CONTRIBUTING.md。

AI 免责声明

这个库的编写得到了大量 AI 协助。核心代码最初由我编写；随着端点 trait 稳定下来，许多端点和文档由 Gemini、CodeRabbit、Claude 和 Codex 扩充，这让添加传输方式的速度大大加快。

生成的代码看起来可能合理，却遗漏重要细节，因此我会手动审查每次提交、清理并重构输出，并依靠集成测试套件发现回归问题。我对当前代码的信任程度，不亚于自己亲自编写的代码。 鉴于功能集庞大，仍可能存在未解决的问题；目前的重点是测试和文档。

许可证

mq-bridge 可由你自行选择，依据 MIT 许可证 或 Apache License, Version 2.0 进行双重许可。

二进制发行版随附适用的第三方条款，详见 THIRD_PARTY_LICENSES.txt。当前维护流程记录在 docs/THIRD_PARTY_LICENSES.md 中。

贡献

除非你明确声明另有约定，否则根据 Apache-2.0 许可证的定义，你有意提交以纳入 mq-bridge 的任何贡献，均按照上述方式进行双重许可，且不附加任何条款或条件。