Go RabbitMQ 简单示例

本文仅分享RabbitMQ在go的简单示例, 具体项目需要更优化的封装

创建RabbitMQ示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
package RabbitMQ

import (
"fmt"
"github.com/streadway/amqp"
"log"
)

// MQUrl 格式 amqp://账号:密码@rabbitmq服务器地址:端口号/vhost
const MQUrl = "amqp://go-test:123456@127.0.0.1:5672/gotest"

type RabbitMQ struct {
conn *amqp.Connection
channel *amqp.Channel
QueueName string // 队列名称
Exchange string // 交换机
Key string // Key
MQUrl string // 连接信息
}

// NewRabbitMQ 创建结构体实例
func NewRabbitMQ(queueName, exchange, key string) *RabbitMQ {
rabbitmq := &RabbitMQ{
QueueName: queueName,
Exchange: exchange,
Key: key,
MQUrl: MQUrl,
}
var err error
rabbitmq.conn, err = amqp.Dial(rabbitmq.MQUrl)
rabbitmq.failOnErr(err, "创建连接错误!")

rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "获取channel失败!")

return rabbitmq
}

// Destroy 断开channel和connection
func (r *RabbitMQ) Destroy() {
_ = r.channel.Close()
_ = r.conn.Close()
}

// failOnErr 错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
if err != nil {
//log.Fatalf("%s:%s", message, err)
panic(fmt.Sprintf("%s:%s", message, err))
}
}

// NewRabbitMQSimple 创建简单模式下的RabbitMQ实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
return NewRabbitMQ(queueName, "", "")
}

// PublishSimple 简单模式下生产消息
func (r *RabbitMQ) PublishSimple(message string) {
// 1. 申请队列, 如果队列不存在会自动创建, 如果存在则跳过创建
// 保证队列存在, 消息能发送到队列中
_, err := r.channel.QueueDeclare(
r.QueueName,
false, // 是否持久化
false, // 是否为自动删除
false, // 是否具有排他性
false, // 是否阻塞
nil, // 额外属性
)
if err != nil {
log.Fatalf("%s:%s", "申请队列失败", err.Error())
}

// 2. 发送消息到队列中
_ = r.channel.Publish(
r.Exchange,
r.QueueName,
false, // 如果为true, 会根据exchange类型和routekey规则,如果无法找到符合条件的队列那么会把发送的消息返回给发送者
false, // 如果为true, 当exchange发送消息到队列后发现队列上没有绑定消费者,则会把消息发还给发送者
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}

// ConsumeSimple 简单模式下消费消息
func (r *RabbitMQ) ConsumeSimple() {
// 1. 申请队列, 如果队列不存在会自动创建, 如果存在则跳过创建
// 保证队列存在, 消息能发送到队列中
_, err := r.channel.QueueDeclare(
r.QueueName,
false, // 是否持久化
false, // 是否为自动删除
false, // 是否具有排他性
false, // 是否阻塞
nil, // 额外属性
)
if err != nil {
log.Fatalf("%s:%s", "申请队列失败", err.Error())
}

// 2. 接收消息
msgs, err := r.channel.Consume(
r.QueueName,
"", // 用来区分多个消费者
true, // 是否自动应答
false, // 是否具有排他性
false, // 如果设置为true,表示不能将同一个connection中发送的消息传递给这个connection中的消费者
false, // 队列消费是否阻塞
nil,
)
if err != nil {
log.Fatalf("%s:%s", "申请队列失败", err.Error())
}

forever := make(chan bool)
go func() {
for msg := range msgs {
log.Printf("Received a message: %s", msg.Body)
fmt.Println(msg.Body)
}
}()
log.Printf("[*] Waiting for message, To exit press CTRL+C")
<-forever
}

建立消费文件和生产文件

  • 消费文件 consume.go
1
2
3
4
5
6
7
8
package main

import "Go/rabbitmq/RabbitMQ"

func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("go-simple")
rabbitmq.ConsumeSimple()
}
  • 生产文件 publish.go
1
2
3
4
5
6
7
8
9
10
11
12
package main

import (
"Go/rabbitmq/RabbitMQ"
"fmt"
)

func main() {
rabbitmq := RabbitMQ.NewRabbitMQSimple("go-simple")
rabbitmq.PublishSimple("Hello Go RabbitMQ!")
fmt.Println("发送成功")
}

Powered by Hexo and Hexo-theme-hiker

Copyright © 2017 - 2022 Keep It Simple And Stupid All Rights Reserved.

访客数 : | 访问量 :