golang-rabbitmq的使用一

golang 可使用库 github.com/streadway/amqp 操作 rabbitmq

安装

go get github.com/streadway/amqp

连接

conn, err := amqp.Dial(amqp://guest:guest@172.17.84.205:5672/)

建立通道

ch, err := conn.Channel()

声明 Queue

q, err := ch.QueueDeclare(
                "testqueue", //Queue name
                true, //durable
                false,
                false,
                false,
                nil,
            )

其中 durable 设为 true 则 queue 持久化，否则不会做持久化。

发布消息

err = ch.Publish(
                "",     //exchange
                q.Name, //routing key(queue name)
                false,
                false,
                amqp.Publishing{
                    DeliveryMode: amqp.Persistent, //Msg set as persistent
                    ContentType:  "text/plain",
                    Body:         []byte(msgBody),
                })

其中 amqp.Publishing 的 DeliveryMode 如果设为 amqp.Persistent 则消息会持久化。
需要注意的是如果需要消息持久化 Queue 也是需要设定为持久化才有效

接收消息

msgs, err := ch.Consume(
                q.Name,
                "MsgWorkConsumer",
                false,  //Auto Ack
                false,
                false,
                false,
                nil,
            )

其中 Auto ack 可以设置为 true。如果设为 true 则消费者一接收到就从 queue 中去除了，如果消费者处理消息中发生意外该消息就丢失了。
如果 Auto ack 设为 false。consumer 在处理完消息后，调用 msg.Ack(false) 后消息才从 queue 中去除。即便当前消费者处理该消息发生意外，只要没有执行 msg.Ack(false) 那该消息就仍然在 queue 中，不会丢失。

生成的 Queue 在生成是设定的参数，下次使用时不能更改设定参数，否则会报错

例子代码如下
conf.go

package config

const (
    RMQADDR     = "amqp://guest:guest@172.17.84.205:5672/"
    QUEUENAME   = "msgQueueWithPersist"
    PRODUCERCNT = 5
    CONSUMERCNT = 20
)

producer.go

package main

import (
    config "xxx/conf"
    "fmt"
    "log"
    "sync"

    "github.com/streadway/amqp"
)

func main() {conn, err := amqp.Dial(config.RMQADDR)
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    var wg sync.WaitGroup
    wg.Add(config.PRODUCERCNT)

    for routine := 0; routine < config.PRODUCERCNT; routine++ {go func(routineNum int) {ch, err := conn.Channel()
            failOnError(err, "Failed to open a channel")
            defer ch.Close()

            q, err := ch.QueueDeclare(
                config.QUEUENAME, //Queue name
                true, //durable
                false,
                false,
                false,
                nil,
            )

            failOnError(err, "Failed to declare a queue")

            for i := 0; i < 500; i++ {msgBody := fmt.Sprintf("Message_%d_%d", routineNum, i)

                err = ch.Publish(
                    "",     //exchange
                    q.Name, //routing key
                    false,
                    false,
                    amqp.Publishing{
                        DeliveryMode: amqp.Persistent, //Msg set as persistent
                        ContentType:  "text/plain",
                        Body:         []byte(msgBody),
                    })

                log.Printf("[x] Sent %s", msgBody)
                failOnError(err, "Failed to publish a message")
            }

            wg.Done()}(routine)
    }

    wg.Wait()

    log.Println("All messages sent!!!!")
}

func failOnError(err error, msg string) {
    if err != nil {fmt.Printf("%s: %s\n", msg, err)
    }
}

consumer.go

package main

import (
    config "xxx/conf"
    "fmt"
    "log"

    "github.com/streadway/amqp"
)

func main() {conn, err := amqp.Dial(config.RMQADDR)
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    forever := make(chan bool)

    for routine := 0; routine < config.CONSUMERCNT; routine++ {go func(routineNum int) {ch, err := conn.Channel()
            failOnError(err, "Failed to open a channel")
            defer ch.Close()

            q, err := ch.QueueDeclare(
                config.QUEUENAME,
                true, //durable
                false,
                false,
                false,
                nil,
            )

            failOnError(err, "Failed to declare a queue")

            msgs, err := ch.Consume(
                q.Name,
                "MsgWorkConsumer",
                false,  //Auto Ack
                false,
                false,
                false,
                nil,
            )

            if err != nil {log.Fatal(err)
            }

            for msg := range msgs {log.Printf("In %d consume a message: %s\n", 0, msg.Body)
                log.Printf("Done")
                msg.Ack(false) //Ack
            }

        }(routine)
    }

    <-forever
}

func failOnError(err error, msg string) {
    if err != nil {fmt.Printf("%s: %s\n", msg, err)
    }
}

完整示例代码在
https://github.com/BinWang-sh…