AMQP Golang优先级不工作

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了AMQP Golang优先级不工作相关的知识,希望对你有一定的参考价值。

我目前正在使用golang(github.com/streadway/amqp)测试rabbitmq,我有三个程序,其中两个程序将消息发送到具有不同优先级的队列和一个从队列中读取的消息。 我遇到的问题是,在用两个程序发送一些消息后,我继续启动将从队列中读取的程序,当它开始从队列中读取时,它会像FIFO一样输出结果。 我希望首先输出高优先级,然后输出优先级较低的消息。 我误解了rabbitmq是如何工作的还是我做错了什么?

发送到队列包主

import (
        "log"

        "github.com/streadway/amqp"
)

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

func main() {
        args := make(amqp.Table)
        args["x-max-priority"] = int64(9)

        conn, err := amqp.Dial("amqp://guest:guest@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(
                "test", // name                                                                                                                                                                                     
                false,   // durable                                                                                                                                                                                 
                false,   // delete when unused                                                                                                                                                                      
                false,   // exclusive                                                                                                                                                                               
                false,   // no-wait                                                                                                                                                                                 
                args,     // arguments                                                                                                                                                                              
        )
        failOnError(err, "Failed to declare a queue")

        body := "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")

        body = "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")

        body = "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")


        body = "high"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 9,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")


        body = "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")


        body = "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")


        body = "low"
        err = ch.Publish(
                "",     // exchange                                                                                                                                                                                 
                q.Name, // routing key                                                                                                                                                                              
                false,  // mandatory                                                                                                                                                                                
                false,  // immediate                                                                                                                                                                                
                amqp.Publishing{
                        Headers:         amqp.Table{},
                        ContentType:     "text/plain",
                        ContentEncoding: "",
                        Body:            []byte(body),
                        DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent                                                                                                                          
                        Priority: 0,
                })
        log.Printf(" [x] Sent %s", body)
        failOnError(err, "Failed to publish a message")

}

阅读队列:

package main

import (
        "log"

        "github.com/streadway/amqp"
)

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

func main() {
        args := make(amqp.Table)
        args["x-max-priority"] = int64(9)

        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")

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

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

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

        defer ch.Close()
        defer conn.Close()

        forever := make(chan bool)

        go func() {
                for d := range msgs {
                        log.Printf("Received a message: %s", d.Body)
                }
        }()

        log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
        <-forever
}

首先,我启动代码向队列发送以下值的消息:

low消息的优先级为0,high的优先级为9。 然后我启动将接收队列的程序。 预期产量:

实际产量:

谢谢

答案

这是我可以从您的代码中找出的,以及您遵循的步骤的描述。

似乎问题是你没有在每段代码中以相同的方式创建队列,低优先级和高优先级的生成器创建没有x-max-priority参数的队列。

由于队列在创建时被定义为优先级队列,并且仅在那时,您实际创建了“标准”队列。

您应该能够轻松确认此检查已创建的队列及其参数。

另一答案

我更新到最新的rabbitmq,它工作,我不知道这是否是唯一的解决方案。

除了调用qazxsw poi之外,我还需要删除qazxsw poi并在之后手动确认该消息,以便它不会确认队列中的所有消息。

以上是关于AMQP Golang优先级不工作的主要内容,如果未能解决你的问题,请参考以下文章

golang使用rabbitmq工作队列

golang使用rabbitmq工作队列

Golang使用amqp发送消息

带有 AMQP 的 Golang 中的 JSON RPC

golang goroutine例子[golang并发代码片段]

golang代码片段(摘抄)