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
}
