Go操作RabbitMQ_go调用rabbitmq

RabbitMQ 常用端口以及作用

                端口   功能       5672   AMQP（Advanced Message Queuing Protocol）协议的默认端口，用于客户端与RabbitMQ服务器之间的通信。       15672   RabbitMQ的管理界面，默认利用HTTP协议，用于监控和管理RabbitMQ服务器。       4369   Erlang分布式节点通信端口，用于RabbitMQ节点之间的通信。       25672   Erlang分布式节点通信端口，用于集群中的内部通信。       5671   安全的AMQP端口，利用TLS/SSL进行加密通信。   架构图

• publisher：生产者，也就是发送消息的一方
  
• consumer：斲丧者，也就是斲丧消息的一方
  
• queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待斲丧者处理
  
• exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
  
• virtual host：假造主机，起到数据隔离的作用。每个假造主机相互独立，有各自的exchange、queue
  

工作模式

1. 简朴模式（Simple）

• 特点：
  
  
• 
  
     
  
  • 生产者直接将消息发送到队列中，斲丧者从队列中获取消息并进行处理。
    
         
    
    • 消息一旦被斲丧，就会从队列中移除。   
      
       
    
    
  
• 真实应用场景：
  
  
• 
  
     
  
  • 电子邮件关照系统：生产者将待发送的邮件放入队列，斲丧者负责从队列中获取邮件并发送。  
    
    
  
• 优点：
  
  
• 
  
     
  
  • 实现简朴，易于明白和部署。
    
         
    
    • 适用于单一斲丧者场景。   
      
       
    
    
  
• 缺点：
  
  
• 
  
     
  
  • 没有消息持久化，一旦 RabbitMQ 服务器重启，队列中的消息将会丢失。  
    
  
  

1.1. 目录结构

simple
-mq
–-rabbitmq.go //这个是RabbitMQ的封装
-SimplePublish
–-mainSimplePublish.go //生产者发送消息
-SimpleRecieve
–-mainSimpleRecieve.go // 斲丧者接受消息

1.2. 代码实现

1.2.1. rabbitmq.go

1. package mq
3. import (
4.     "fmt"
5.     "log"
7.     "github.com/streadway/amqp"
8. )
10. // MQURL amqp://user:password@host:port/vhost
11. // amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host
12. // 如果vhost是“/”就输入/%2F，/%2F代表斜杠
13. const MQURL = "amqp://test:123123@127.0.0.1:5672/%2F"
15. // RabbitMQ rabbitMQ结构体
16. type RabbitMQ struct {
17.     conn    *amqp.Connection
18.     channel *amqp.Channel
19.     //队列名称
20.     QueueName string
21.     //交换机名称
22.     Exchange string
23.     //bind Key 名称
24.     Key string
25.     //连接信息
26.     Mqurl string
27. }
29. // NewRabbitMQ 创建结构体实例
30. func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
31.     return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}
32. }
34. // Destory 断开channel 和 connection
35. func (r *RabbitMQ) Destory() {
36.     err := r.channel.Close()
37.     if err != nil {
38.         return
39.     }
40.     err = r.conn.Close()
41.     if err != nil {
42.         return
43.     }
44. }
46. // 错误处理函数
47. func (r *RabbitMQ) failOnErr(err error, message string) {
48.     if err != nil {
49.         log.Fatalf("%s:%s", message, err)
50.     }
51. }
53. // NewRabbitMQSimple 创建简单模式下RabbitMQ实例
54. func NewRabbitMQSimple(queueName string) *RabbitMQ {
55.     //创建RabbitMQ实例
56.     rabbitmq := NewRabbitMQ(queueName, "", "")
57.     var err error
58.     //获取connection
59.     rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
60.     rabbitmq.failOnErr(err, "failed to connect rabb"+
61.                        "itmq!")
62.     //获取channel
63.     rabbitmq.channel, err = rabbitmq.conn.Channel()
64.     rabbitmq.failOnErr(err, "failed to open a channel")
65.     return rabbitmq
66. }
68. // PublishSimple simple模式队列生产
69. func (r *RabbitMQ) PublishSimple(message string) {
70.     //1.申请队列，如果队列不存在会自动创建，存在则跳过创建
71.     _, err := r.channel.QueueDeclare(
72.         // queue:队列名称
73.         r.QueueName,
74.         //durable:是否持久化，当mq重启之后，还在
75.         true,
76.         //exclusive:是否独占即只能有一个消费者监听这个队列
77.         false,
78.         //autoDelete:是否自动删除。当没有Consumer时，自动删除掉
79.         false,
80.         //noWait:是否阻塞处理。true:不阻塞，false:阻塞
81.         false,
82.         //arguments:其他属性
83.         nil,
84.     )
85.     if err != nil {
86.         fmt.Println(err)
87.     }
88.     //调用channel 发送消息到队列中
89.     r.channel.Publish(
90.         r.Exchange,
91.         r.QueueName,
92.         //如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
93.         false,
94.         //如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
95.         false,
96.         amqp.Publishing{
97.             ContentType: "text/plain",
98.             Body:        []byte(message),
99.         })
100. }
102. // ConsumeSimple simple模式下消费者
103. func (r *RabbitMQ) ConsumeSimple() {
104.     //1.申请队列，如果队列不存在会自动创建，存在则跳过创建
105.     q, err := r.channel.QueueDeclare(
106.         //队列名称
107.         r.QueueName,
108.         //是否持久化
109.         true,
110.         //是否自动删除
111.         false,
112.         //是否具有排他性
113.         false,
114.         //是否阻塞处理
115.         false,
116.         //额外的属性
117.         nil,
118.     )
119.     if err != nil {
120.         fmt.Println(err)
121.     }
122.     //接收消息
123.     msgs, err := r.channel.Consume(
124.         q.Name, // queue
125.         //用来区分多个消费者
126.         "", // consumer
127.         //是否自动应答
128.         true, // auto-ack
129.         //是否独有
130.         false, // exclusive
131.         //设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
132.         false, // no-local
133.         //列是否阻塞
134.         false, // no-wait
135.         nil,   // args
136.     )
137.     if err != nil {
138.         fmt.Println(err)
139.     }
141.     forever := make(chan bool)
142.     //启用协程处理消息
143.     go func() {
144.         for d := range msgs {
145.             //消息逻辑处理，可以自行设计逻辑
146.             log.Printf("Received a message: %s", d.Body)
148.         }
149.     }()
151.     log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
152.     <-forever
154. }

复制代码

1.2.2. SimplePublish.go

1. package main
3. import (
4.     "fmt"
5.     "qiutian.com/study/02.rabbitmq/Simple/mq"
6. )
8. func main() {
9.     rabbitmq := mq.NewRabbitMQSimple("hello.queue1")
10.     rabbitmq.PublishSimple("Hello akita!")
11.     fmt.Println("发送成功！")
12. }

复制代码

1.2.3. mainSimpleRecieve.go

1. <

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。