---
title: RabbitMQ Source 集成
---

# RabbitMQ Source 集成

本文介绍如何在管道的 Source 步骤中使用 RabbitMQ 作为数据来源。

## 适用场景

- 接入已有 AMQP 生产链路数据
- 从交换机/队列读取事件并汇入 FlowMQ

## 前提条件

- 已了解管道结构与创建流程，参考 [数据集成](data-integration.md)；Source 连接器总览见 [Source](data-integration-input.md)
- RabbitMQ 服务可访问
- 已具备目标 vhost 与队列访问权限

## 配置步骤

1. 在“数据管道”中创建管道，完成基础信息配置。
2. 在 Source 步骤选择 `RabbitMQ`。
3. 配置 `urls`、`queue`、`consumer_tag` 等参数。
4. 点击“测试”验证连接。
5. 继续完成 Processors、Sink 与确认步骤。

## 配置示例

可参考以下参数示例：

| 参数 | 建议值 |
|---|---|
| host | `127.0.0.1` |
| port | `5672` |
| username | `guest` |
| password | `guest` |
| vhost | `/` |
| queue | `flowmq.source.rabbitmq` |
| consumer_tag | `flowmq-source-1` |
| queue_declare.enabled | `true` |
| queue_declare.durable | `false` |
| auto_ack | `true` |
| tls.enabled | `false` |

## 必要表单参数

以表单中带 `*` 的字段为必填：

| 参数 | 说明 |
|------|------|
| `urls*` | 连接地址列表（每项包含 host/port，可含用户名、密码、vhost） |
| `queue*` | 要消费的 AMQP 队列名 |
| `consumer_tag*` | 消费者标识。用于在 RabbitMQ 侧区分本管道的消费连接；控制台要求必填，建议填可读且唯一的名称（如 `flowmq-source-1`） |
| `auto_ack*` | 是否在消费时自动确认。开启可提高吞吐，但会削弱投递保证 |
| `prefetch_count*` | 未确认消息的最大条数 |
| `prefetch_size*` | 未确认消息的最大字节数（`0` 表示不按字节限制） |

可选高级项：

| 参数 | 说明 |
|------|------|
| `tls` | 自定义 TLS |
| `queue_declare` | 被动声明目标队列 |
| `bindings_declare` | 被动声明队列绑定 |

## 参数建议

- 优先使用独立消费队列避免互相干扰
- `consumer_tag` 同一连接上不要与其他消费者重复
- 按业务类型规划 routing key

## 注意事项

- 明确 ack 策略，避免消息堆积
- 关注重试与死信策略
- 接入后的字段清洗与映射见 [Processors](data-integration-processors-expressions.md)
- 控制台 `urls` 按 host/port 对象填写（可含 username、password、vhost）；Bento 按该对象格式解析，不要改成 `amqp://...` 字符串
- 使用 FlowMQ 内置 AMQP（`--class=amqp`）时，建议开启 `queue_declare.enabled`，并将 `queue_declare.durable` 设为 `false`（内置服务不支持 durable queue）
- 队列名不要与已有 Kafka Topic / FlowMQ 流同名（例如不要用 `flowmq.mqtt.kafka`），否则 `queue_declare` 可能返回 406
- 非 TLS 场景下请将 `tls.enabled` 保持为 `false`；若误开 TLS，连通性测试可能仍通过，但管道任务无法正常运行
- Sink 为 Drop 的 Source 测试场景，建议开启 `auto_ack`，便于观察处理计数