Golang中RabbitMQ通道在消息数超过QoS限制时冻结的解决方案

Golang中RabbitMQ通道在消息数超过QoS限制时冻结的解决方案 我在排查Go应用程序中遇到的RabbitMQ问题时，偶然看到了这篇帖子（网站 在通道上使用QoS时，如果消息数量超过QoS定义的消息数量，通道会停止）。问题是，当我在通道上使用QoS并且消息数量超过定义的限制时，该通道会停止接收消息。

以下是我正在使用的代码：

package rmqcode

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Panicf("%s: %s", msg, err)
    }
}

func recWorker(worknum int, del *amqp.Delivery) {
    println(fmt.Sprintf("start %d", worknum))
    time.Sleep(time.Second * 2)
    println(fmt.Sprintf("end %d", worknum))
    del.Ack(true)
}

func Receiver() {
    conn, err := amqp.Dial("amqp://user:pass@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "hello", // name
        true,    // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

    err = ch.Qos(
        3,     // prefetch count
        0,     // prefetch size
        false, // global
    )
    failOnError(err, "Failed to set QoS")

    msg, err := ch.Consume(
        q.Name, // queue
        "",     // consumer
        false,  // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failOnError(err, "Failed to register a consumer")

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        var counter = 1
        for msgs := range msg {
            go recWorker(counter, &msgs)
            counter = counter + 1
        }
        fmt.Println("stopped")
        wg.Done()
    }()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    wg.Wait()
}

func Send() {
    conn, err := amqp.Dial("amqp://user:pass@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "hello", // name
        true,    // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    body := "Hello World!"
    err = ch.PublishWithContext(ctx,
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,  // immediate
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(body),
        })
    failOnError(err, "Failed to publish a message")
    log.Printf(" [x] Sent %s\n", body)
}

在我的情况下，我将QoS设置为一个特定的数字，但当消息数量超过这个数字时，消费者就不再处理后续的消息。我已经尝试按照帖子中所示运行代码，但仍然面临同样的问题。我怀疑这与消息的确认方式或预取计数有关，但我不完全确定。

有没有其他人在使用RabbitMQ和QoS时遇到过这个问题？对于如何解决这个问题或如何正确处理消息队列而不导致通道冻结，有什么建议吗？