golang rabbitmq的使用（二）

上一篇文章中主要说明了关于queue的使用。producer和consumer都是直接连接到queue进行消息的生产和消费。本篇主要说明Exchange的使用。

Exchange有direct,topic,headers和fanout几种类型
先说一下fanout类型的exchange。作用就是将消息转发到所有和他连接的queue。所有producer可以不用关心有多少queue或者consumer，只需将消息发送到exchange，至于说哪些consumer需要这些消息，直接将queue连接到exchange就可以获取想要的消息。

exchange的定义如下

err = ch.ExchangeDeclare(
  "fanout_exchange",   // name
  "fanout", // type
  true,     // durable
  false,    // auto-deleted
  false,    // internal
  false,    // no-wait
  nil,      // arguments
)

定义与exchange连接的queue

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

此处queue的name没有指定，生成的queue的名字是随机的
exclusive为true，当连接关闭时该queue会被删除

将queue与exchange连接在一起的过程成为binding，用法如下

err = ch.QueueBind(
  q.Name, // queue name
  "",     // routing key
  "fanout_exchange", // exchange
  false,
  nil,
)

例子代码如下
conf.go

package config

const (
    RMQADDR      = "amqp://guest:guest@172.17.84.205:5672/"
    EXCHANGENAME = "fanout_exchange"
    CONSUMERCNT  = 3
)

producer.go

package main

import (
    config "binTest/rabbitmqTest/t1/l3/conf"
    "fmt"
    "log"
    "os"

    "github.com/streadway/amqp"
)

func main() {

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

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

    err = ch.ExchangeDeclare(
        "fanout_exchange", //exchange name
        "fanout",          //exchange kind
        true,              //durable
        false,             //autodelete
        false,
        false,
        nil,
    )

    failOnError(err, "Failed to declare exchange")

    msgs := os.Args[1:]
    msgNum := len(msgs)

    for cnt := 0; cnt < msgNum; cnt++ {
        msgBody := msgs[cnt]
        err = ch.Publish(
            "fanout_exchange", //exchange
            "",                //routing key
            false,
            false,
            amqp.Publishing{
                ContentType: "text/plain",
                Body:        []byte(msgBody),
            })

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

}

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

consumer.go

package main

import (
    config "binTest/rabbitmqTest/t1/l3/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)

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

    err = ch.ExchangeDeclare(
        "fanout_exchange", //exchange name
        "fanout",          //exchange kind
        true,              //durable
        false,             //autodelete
        false,
        false,
        nil,
    )

    failOnError(err, "Failed to declare exchange")

    for routine := 0; routine < config.CONSUMERCNT; routine++ {
        go func(routineNum int) {

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

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

            err = ch.QueueBind(
                q.Name,
                "",
                "fanout_exchange",
                false,
                nil,
            )
            failOnError(err, "Failed to bind exchange")

            msgs, err := ch.Consume(
                q.Name,
                "",
                true, //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", routineNum, msg.Body)
            }

        }(routine)
    }

    <-forever
}

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

执行结果如下
producer

consumer

在exchange上binding了三个queue，producer发送的每个消息都被送到了三个queue中被三个不同的consumer消费

代码在
[https://github.com/BinWang-sh...]

有疑问加站长微信联系（非本文作者）