Rabbitmq Endpoint
Rabbitmq Endpoint v0.24.0+ 用来创建和启动Rabbitmq订阅服务,它可以订阅不同流名称数据,然后路由到不同规则链进行处理。
提示
- 该组件是扩展组件,需要引入额外的扩展库:rulego-components (opens new window)
# Type
endpoint/rabbitmq
# 启动配置
该组件允许通关过server字段复用共享的连接客户端。参考组件连接复用 。
| 字段 | 类型 | 说明 | 默认值 |
|---|---|---|---|
| server | string | RabbitMQ服务器地址,示例:amqp://guest:guest@127.0.0.1:5672/ | 无 |
| exchange | string | 交换机名称 | 无 |
| exchangeType | string | 交换机类型,可选值:direct, fanout, topic | 无 |
| durable | bool | 表示交换器和队列是否持久化。如果设置为 true,即使消息服务器重启,也会被保留。 | false |
| autoDelete | bool | 表示队列是否自动删除。如果设置为 true,则当最后一个消费者断开后,队列会被自动删除。 | false |
| queueName | string | 自定义队列名称。不设置时使用自动生成的匿名独占队列,连接断开即删除(历史行为);设置固定名称(配合durable=true、autoDelete=false)后,消费者离线或消息服务器重启期间消息会堆积在队列,消费者重新上线后继续消费 | 无(匿名队列) |
| exclusive | bool | 队列是否仅限当前连接使用,连接断开即删除。queueName为空时强制为true;具名队列默认false | false |
提示
要实现「broker 重启、消费者离线都不丢消息,消费者上线后继续消费」,配置组合为:queueName 固定名称 + durable=true + autoDelete=false,且发布方消息须为持久化消息(delivery mode 2,例如发布端组件 x/rabbitmqClient 开启 persistent)。同名队列被多次声明时参数必须一致,否则声明报错。
# 响应
exchange.Out.SetBody响应之前,需要通过exchange.Out.Headers()或者exchange.Out.Msg.Metadata指定responseTopic参数,组件就会往指定的主题发送数据:
exchange.Out.GetMsg().Metadata.PutValue("responseTopic", "device.msg.response")
// or
exchange.Out.Headers().Add("responseTopic", "device.msg.response")
exchange.Out.SetBody([]byte("ok"))
1
2
3
4
5
2
3
4
5
响应参数配置:
| 字段 | 类型 | 是否必填 | 说明 | 默认值 |
|---|---|---|---|---|
| responseTopic/responseKey | string | 是 | 响应Key | - |
| responseExchange | string | 是 | 响应交换机 | - |
# 多副本部署
本端点为每个副本声明独立的私有队列绑定同一交换机,多副本部署时每条消息会触发所有副本。规则链配置注入分布式锁后,本端点自动单活选主:只有主副本声明队列并消费,主副本失联后待命副本自动接管。单机部署不受影响。
配置固定queueName(非独占)时,各副本声明并消费同一个具名队列:未配置分布式锁时多副本以竞争消费者模式轮询分摊消息;配置分布式锁时依旧单活消费,主副本切换后待命副本从队列继续消费,切换窗口内发布的消息不会丢失。
# 示例
以下是使用endpoint的示例代码:
在 GitHub 上编辑此页 (opens new window)
上次更新: 2026/10/13, 03:03:03