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文件目录执行
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 }
有疑问加站长微信联系(非本文作者)