由message绑定queue到exchange
This commit is contained in:
@@ -84,6 +84,7 @@ func (c *Consumer) Start(ctx context.Context) (err error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 声明队列(如果不存在则创建)
|
// 声明队列(如果不存在则创建)
|
||||||
|
// 注意:Queue到Exchange的绑定应由message服务在发送响应时动态创建,或通过运维工具提前配置
|
||||||
_, err = ch.QueueDeclare(
|
_, err = ch.QueueDeclare(
|
||||||
c.queue, // name
|
c.queue, // name
|
||||||
true, // durable(持久化)
|
true, // durable(持久化)
|
||||||
@@ -96,21 +97,6 @@ func (c *Consumer) Start(ctx context.Context) (err error) {
|
|||||||
return gerror.Newf("声明队列失败: %v", err)
|
return gerror.Newf("声明队列失败: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: 队列绑定逻辑暂时注释,避免重复binding导致消息重复投递
|
|
||||||
// 绑定队列到Exchange(使用队列名作为routingKey,支持多租户)
|
|
||||||
// Exchange类型应该是topic,绑定模式为 #(接收所有消息)
|
|
||||||
// err = ch.QueueBind(
|
|
||||||
// c.queue, // queue name
|
|
||||||
// "#", // routing key(通配符,接收所有消息)
|
|
||||||
// "ragflow.response", // exchange name
|
|
||||||
// false, // noWait
|
|
||||||
// nil, // arguments
|
|
||||||
// )
|
|
||||||
// if err != nil {
|
|
||||||
// g.Log().Warningf(ctx, "绑定队列到Exchange失败(可能Exchange不存在或类型不匹配): %v", err)
|
|
||||||
// // 不返回错误,继续启动消费者(可能是direct exchange或队列已绑定)
|
|
||||||
// }
|
|
||||||
|
|
||||||
// 设置 QoS(并发控制)
|
// 设置 QoS(并发控制)
|
||||||
err = ch.Qos(
|
err = ch.Qos(
|
||||||
c.prefetchCount, // prefetchCount: 每个 consumer 最多同时处理的消息数
|
c.prefetchCount, // prefetchCount: 每个 consumer 最多同时处理的消息数
|
||||||
|
|||||||
Reference in New Issue
Block a user