AMQP

本文介绍如何将 FlowMQ 数据实时写入 AMQP。

前提条件

• 已了解管道结构与创建流程，参考 数据集成；Sink 连接器总览见 Sink
• AMQP 服务可访问
• 已具备交换机/队列写入权限

配置步骤

1. 在“数据管道”中创建管道，完成基础信息配置。
2. 在 Source 步骤选择 Kafka（Topic：flowmq.mqtt.kafka），Processors 按需配置。
3. 在 Sink 步骤选择 AMQP，配置 urls、exchange、key 等参数。
4. 点击“测试”验证连接。
5. 继续完成确认步骤。

配置示例

以下示例将 Kafka Topic flowmq.mqtt.kafka 中的样例消息写入目标系统。样例 JSON 与 Source 约定见 Sink · 示例场景。

可参考以下参数示例：

参数	建议值
urls	host 127.0.0.1，port 5672，username/password guest，vhost /
exchange	amq.topic
key	flowmq.sink.demo
max_in_flight	64

必要表单参数

参数名后带 * 表示控制台必填。「支持表达式」为「是」时，可使用 ${! ... } 按消息动态取值，详见 Processors。

参数	支持表达式	说明
urls*	否	连接地址列表（每项包含 host/port，可含用户名、密码、vhost）
exchange*	是	目标交换机
max_in_flight*	否	并发写入上限
key	是	routing key（按下游绑定规则填写，与 Kafka Topic 名无关）

参数建议

• exchange、key 需与下游绑定规则一致；也可按消息动态生成
• max_in_flight 可按目标吞吐逐步调高

注意事项

• key 为 AMQP routing key，须与队列绑定规则一致
• 可按实际吞吐调整 max_in_flight