分享
  1. 首页
  2. 文章

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

shi-meng · · 15448 次点击 · · 开始浏览
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

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实践 (二 实现简单的消息收发) 

					

					
				

				
				

	
入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

	
		关注微信
	


				

					
15448 次点击 

					
					
					
				

			

			
			
			

				

					
					
				

			

			
			
			
			
			

				

					
					
						2 回复 | 直到 2025年05月15日 08:50:44
					
				

				
				

					
暂无回复

				

				
			

			
			
			



	
	
添加一条新回复
	
	
	(您需要 后才能回复  没有账号 ?)
	
	

	

		
		

			

				
			

			
			

				
    
					
  • 请尽量让自己的回复能够对别人有帮助
    • 
					
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
    • 
					
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet 
    • 
					
  • 图片支持拖拽、截图粘贴等方式上传
    • 
				

				

			

		

	



			
			
		

	

	

		
		

	
	

		
 用户登录

	

	

		

			

				

					
				

			

			

				

					
				

			

			

				

					

						
					

					
					

						
没有账号?注册 

						
					

					
					

						
						或
						
					

					
					
					
					
				

			

		

	

	



		
		

			
				
(追記) (追記ここまで)
			
		

		
		

	

		
 今日阅读排行

	

	

		

			
    
				加载中
			

		

	



		
		

			
				
(追記) (追記ここまで)
			
		

		
		

	

		
 一周阅读排行

	

	

		

			
    
				加载中
			

		

	



		

 

 
 关注我

 

 

 

 
    
 
  • 
 扫码关注领全套学习资料
 关注微信公众号
 
    • 
 
  • 
 加入 QQ 群:
 
      
 
    • 192706294(已满)
      • 
 
    • 731990104(已满)
      • 
 
    • 798786647(已满)
      • 
 
    • 729884609(已满)
      • 
 
    • 977810755(已满)
      • 
 
    • 815126783(已满)
      • 
 
    • 812540095(已满)
      • 
 
    • 1006366459(已满)
      • 
 
    • 692541889
      • 
 
    
 
    
 
    • 
								
  • 
 关注微信公众号
								
    • 
 
  • 
 加入微信群:liuxiaoyan-s,备注入群
 
    • 
 
  • 也欢迎加入知识星球 Go粉丝们(免费) 
    • 
 

 

 



	





 

 

 

 
 

 给该专栏投稿
 写篇新文章
 

 每篇文章有总共有 5 次投稿机会
 
 

 

 
 

 
 
 

 

 
 

 





 

 

 

 
 
收入到我管理的专栏
 新建专栏