1. 首页
  2. 文章

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

shi-meng ·

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
}




					

					
					
					

						
本文来自:博客园 

						
感谢作者:shi-meng

						
查看原文:golang rabbitmq实践 (二 实现简单的消息收发) 

					

					
				

				

					
				

				
			

			
			
			
			
			

				

					
					
前辈,你代码有两处错误 在Push方法中的channel.Publish(exchange, queueName, false, false, amqp.Publishing{......}中的queueName应该是交换机与队列相连的binkey,receive方法中的 msgs, err := channel.Consume(queueName, &#34;&#34;, true, false, false, false, nil)中的queueName才是队列名 

					
					#1
				

			

			
			

				
				
更多评论

				
				
				

					
					
Great! [balance bike](https://www.xjd.com/c-Kids-Balance-Bike/)

					
					#2
				

				
				
			

		

	

	

		
		

	
	

		
 用户登录

	

	

		

			

				

					
				

			

			

				

					
				

			

			

				

					

						
					

					
					

						
没有账号?注册 

						
					

					
					

						
						或
						
					

					
					
					
					
				

			

		

	

	





	
		
	


		
		

	

		
 今日阅读排行

	

	

		

			
    
				加载中
			

		

	



		
		
		

	

		
 一周阅读排行

	

	

		

			
    
				加载中