Go操作RabbitMQ_go调用rabbitmq - ToB企服应用市场:ToB评测及商务社交产业平台

package mq
import (
"fmt"
"log"
"github.com/streadway/amqp"
)
// MQURL amqp://user:password@host:port/vhost
// amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host
// 如果vhost是“/”就输入/%2F，/%2F代表斜杠
const MQURL = "amqp://test:123123@127.0.0.1:5672/%2F"
// RabbitMQ rabbitMQ结构体
type RabbitMQ struct {
conn *amqp.Connection
channel *amqp.Channel
//队列名称
QueueName string
//交换机名称
Exchange string
//bind Key 名称
Key string
//连接信息
Mqurl string
}
// NewRabbitMQ 创建结构体实例
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}
}
// Destory 断开channel 和 connection
func (r *RabbitMQ) Destory() {
err := r.channel.Close()
if err != nil {
return
}
err = r.conn.Close()
if err != nil {
return
}
}
// 错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
if err != nil {
log.Fatalf("%s:%s", message, err)
}
}
// NewRabbitMQSimple 创建简单模式下RabbitMQ实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
//创建RabbitMQ实例
rabbitmq := NewRabbitMQ(queueName, "", "")
var err error
//获取connection
rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
rabbitmq.failOnErr(err, "failed to connect rabb"+
"itmq!")
//获取channel
rabbitmq.channel, err = rabbitmq.conn.Channel()
rabbitmq.failOnErr(err, "failed to open a channel")
return rabbitmq
}
// PublishSimple simple模式队列生产
func (r *RabbitMQ) PublishSimple(message string) {
//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
_, err := r.channel.QueueDeclare(
// queue:队列名称
r.QueueName,
//durable:是否持久化，当mq重启之后，还在
true,
//exclusive:是否独占即只能有一个消费者监听这个队列
false,
//autoDelete:是否自动删除。当没有Consumer时，自动删除掉
false,
//noWait:是否阻塞处理。true:不阻塞，false:阻塞
false,
//arguments:其他属性
nil,
)
if err != nil {
fmt.Println(err)
}
//调用channel 发送消息到队列中
r.channel.Publish(
r.Exchange,
r.QueueName,
//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
false,
//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
})
}
// ConsumeSimple simple模式下消费者
func (r *RabbitMQ) ConsumeSimple() {
//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
q, err := r.channel.QueueDeclare(
//队列名称
r.QueueName,
//是否持久化
true,
//是否自动删除
false,
//是否具有排他性
false,
//是否阻塞处理
false,
//额外的属性
nil,
)
if err != nil {
fmt.Println(err)
}
//接收消息
msgs, err := r.channel.Consume(
q.Name, // queue
//用来区分多个消费者
"", // consumer
//是否自动应答
true, // auto-ack
//是否独有
false, // exclusive
//设置为true，表示不能将同一个Conenction中生产者发送的消息传递给这个Connection中的消费者
false, // no-local
//列是否阻塞
false, // no-wait
nil, // args
)
if err != nil {
fmt.Println(err)
}
forever := make(chan bool)
//启用协程处理消息
go func() {
for d := range msgs {
//消息逻辑处理，可以自行设计逻辑
log.Printf("Received a message: %s", d.Body)
}
}()
log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever
}

复制代码

package main
import (
"fmt"
"qiutian.com/study/02.rabbitmq/Simple/mq"
)
func main() {
rabbitmq := mq.NewRabbitMQSimple("hello.queue1")
rabbitmq.PublishSimple("Hello akita!")
fmt.Println("发送成功！")
}

复制代码

<

复制代码