订阅模式

Publish/Subscribe,订阅模式

消息被路由投递给多个队列，一个消息可以被多个消费者获取

订阅模式代码

package RabbitMQ

import (
    "fmt"
    "github.com/streadway/amqp"
    "log"
)

const MQURL = "amqp://guest:guest@127.0.0.1:5672/imooc"

//创建rabbitmq结构体实例
type RabbitMQ struct {
    conn      *amqp.Connection
    channel   *amqp.Channel
    QueueName string
    Exchange  string
    Key       string
    Mqurl     string
}

func NewRabbitMQ(queueName string, Exchange string, key string) *RabbitMQ {
    rabbitmq := &RabbitMQ{QueueName: queueName, Exchange: Exchange, Key: key, Mqurl: MQURL}
    var err error
    rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
    rabbitmq.failOnErr(err, "创建连接错误")
    rabbitmq.channel, err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err, "获取channel失败")
    return rabbitmq
}

//断开channel和connection
func (r *RabbitMQ) Destroy() {
    r.channel.Close()
    r.conn.Close()
}

//错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
    if err != nil {
        log.Fatalf("%s:%s", message, err)
        panic(fmt.Sprintf("%s%s", message, err))
    }
}

//订阅模式创建RabbitMQ实例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
    rabbitmq := NewRabbitMQ("", exchangeName, "")
    var err error
    rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
    rabbitmq.failOnErr(err, "创建连接错误")
    rabbitmq.channel, err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err, "获取channel失败")
    return rabbitmq
}

//订阅模式生产
func (r *RabbitMQ) PublishPub(message string) {
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        //广播类型
        "fanout",
        true,
        false,
        //Yes表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
        false,
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare an excha"+"nge")
    //发送消息
    err = r.channel.Publish(
        r.Exchange,
        "",
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(message),
        })
}
func (r *RabbitMQ) ReceiveSub() {
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        //交换机类型
        "fanout",
        true,
        false,
        //Yes表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
        false,
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare an exchange")
    q, err := r.channel.QueueDeclare(
        "", // 随机生产队列名称
        false,
        false,
        true,
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare a queue")
    //绑定队列到exchange中
    err = r.channel.QueueBind(
        q.Name,
        //在Pub/sub模式下，这里的key要为空
        "",
        r.Exchange,
        false,
        nil,
    )
    //消费消息
    messages, err := r.channel.Consume(
        q.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    forever := make(chan bool)
    go func() {
        for d := range messages {
            log.Printf("Received a message:%s", d.Body)
        }
    }()
    fmt.Println("推出请按CTRL+C
")
    <-forever
}

订阅模式生产者代码

package main

import (
    "immoc-rabbitmq/RabbitMQ"
    "strconv"
    "time"
)

func main() {
    rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
    for i := 0; i < 100; i++ {
        rabbitmq.PublishPub("订阅模式生产第" + strconv.Itoa(i) + "条数据")
        time.Sleep(time.Second)
    }
}

订阅模式消费者代码

package main

import "immoc-rabbitmq/RabbitMQ"

func main() {
    rabbitmq := RabbitMQ.NewRabbitMQPubSub("newProduct")
    rabbitmq.ReceiveSub()
}

来自为知笔记(Wiz)