golang rabbitmq實踐 (二 實現簡單的消息收發)

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
}