golang rabbitmq实践 (二 实现简单的消息收发)

知识标签: golang
1:驱动

  本来打算自己写一个驱动的,后来发现github上面已经有了,那我就直接拿现成的了, 驱动采用 github.com/streadway/amqp ,直接import就可以啦!
 2:exchange and queue

  在上一篇文章中,我们已经创建好virtualhost 、exchange and queue,所以我们先定义这些常量

  

const (
    queueName = "push.msg.q"
    exchange  = "t.msg.ex"
    mqurl ="amqp://shi:123@192.168.232.130:5672/test"
)
var conn *amqp.Connection
var channel *amqp.Channel

 

3:错误处理

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

4:连接mq

func mqConnect() {
    var err error
    conn, err = amqp.Dial(mqurl)
    failOnErr(err, "failed to connect tp rabbitmq")

    channel, err = conn.Channel()
    failOnErr(err, "failed to open a channel")
}


 
5:push

  先上代码:

func push() {

    if channel == nil {
        mqConnect()
    }
    msgContent := "hello world!"

    channel.Publish(exchange, queueName, false, false, amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(msgContent),
    })
}


  其实是很简单的,调用 channel函数的Publish方法,传入exchange name 和 queue name,最后一个参数是消息内容,ContentType我们设置为text/plain, 为文本类型,body是消息内容,要传入字节数组,这样就完成了一条消息的push,接下来我们再看receive
6:receive

 代码:

func receive() {
    if channel == nil {
        mqConnect()
    }

    msgs, err := channel.Consume(queueName, "", true, false, false, false, nil)
    failOnErr(err, "")

    forever := make(chan bool)

    go func() {
        //fmt.Println(*msgs)
        for d := range msgs {
            s := BytesToString(&(d.Body))
            count++
            fmt.Printf("receve msg is :%s -- %d\n", *s, count)
        }
    }()

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


  通过调用channel.Consume函数返回一个接受消息的chan类型管道,然后range 这个chan,接收到的数据是[]byte,转换为string后输出

  <-forever  这个是为了控制当前线程不退出
7:入口main

  

func main() {
    go func() {
        for {
            push()
            time.Sleep(1 * time.Second)
        }
    }()
    receive()
    fmt.Println("end")
    close()
}


  for 循环保证每秒发送一条消息到mq,这个地方采用协程保证不阻塞主线程。receive函数不能采用协程,不然主线程就退出了。close函数是释放连接对象,但是在这个例子中是没有起效的,因为线程永远都不会自动退出,只能认为的CTRL+C 或者程序死掉,系统重启

 
8:执行:

切换到go文件目录执行
go run main.go

//运行日志

receve msg is :hello world! -- 1246
receve msg is :hello world! -- 1247
receve msg is :hello world! -- 1248
receve msg is :hello world! -- 1249
receve msg is :hello world! -- 1250
receve msg is :hello world! -- 1251
receve msg is :hello world! -- 1252
receve msg is :hello world! -- 1253
receve msg is :hello world! -- 1254
receve msg is :hello world! -- 1255
receve msg is :hello world! -- 1256
receve msg is :hello world! -- 1257
receve msg is :hello world! -- 1258
receve msg is :hello world! -- 1259
receve msg is :hello world! -- 1260
receve msg is :hello world! -- 1261
receve msg is :hello world! -- 1262
receve msg is :hello world! -- 1263
receve msg is :hello world! -- 1264
receve msg is :hello world! -- 1265
receve msg is :hello world! -- 1266

9:全部代码

package main

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

var conn *amqp.Connection
var channel *amqp.Channel
var count = 0

const (
    queueName = "push.msg.q"
    exchange  = "t.msg.ex"
    mqurl ="amqp://shi:123@192.168.232.130:5672/test"
)

func main() {
    go func() {
        for {
            push()
            time.Sleep(1 * time.Second)
        }
    }()
    receive()
    fmt.Println("end")
    close()
}

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

func mqConnect() {
    var err error
    conn, err = amqp.Dial(mqurl)
    failOnErr(err, "failed to connect tp rabbitmq")

    channel, err = conn.Channel()
    failOnErr(err, "failed to open a channel")
}

func close() {
    channel.Close()
    conn.Close()
}

//连接rabbitmq server
func push() {

    if channel == nil {
        mqConnect()
    }
    msgContent := "hello world!"

    channel.Publish(exchange, queueName, false, false, amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(msgContent),
    })
}

func receive() {
    if channel == nil {
        mqConnect()
    }

    msgs, err := channel.Consume(queueName, "", true, false, false, false, nil)
    failOnErr(err, "")

    forever := make(chan bool)

    go func() {
        //fmt.Println(*msgs)
        for d := range msgs {
            s := BytesToString(&(d.Body))
            count++
            fmt.Printf("receve msg is :%s -- %d\n", *s, count)
        }
    }()

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

func BytesToString(b *[]byte) *string {
    s := bytes.NewBuffer(*b)
    r := s.String()
    return &r
}
沪ICP备09053415号 © 赶知识网 我的题集