diff --git a/rabbitmq.go b/rabbitmq.go index 382aa90..a933f8f 100644 --- a/rabbitmq.go +++ b/rabbitmq.go @@ -46,13 +46,19 @@ func (r *rabbitmq) Publish(exchangeName, msg string, delay ...int64) error { // Consume @Title 消费队列 // @Param queueName string true "队列名称" // @Param autoAck bool true "是否自动确认" +// @Param prefetchCount int 0 "预消费数量" // @Param callback func(amqp.Delivery)) true "回调处理" -func (r *rabbitmq) Consume(queueName string, autoAck bool, callback func(amqp.Delivery)) error { +func (r *rabbitmq) Consume(queueName string, autoAck bool, prefetchCount int, callback func(amqp.Delivery)) error { ch, err := r.Conn.Channel() if err != nil { return err } defer ch.Close() + if prefetchCount > 0 { + if err := ch.Qos(prefetchCount, 0, false); err != nil { + return err + } + } //消费队列 msgs, err := ch.Consume( queueName, //队列名称