Go RabbitMQ 简单示例

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

创建RabbitMQ示例

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

package main

import "Go/rabbitmq/RabbitMQ"

func main() {
    rabbitmq := RabbitMQ.NewRabbitMQSimple("go-simple")
    rabbitmq.ConsumeSimple()
}

• 生产文件 publish.go

package main

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

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