beego RabbitMQ 简单模式 工作模式

MQ服务  

package mq

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

// 连接
func Connect() (*amqp.Connection, error) {
	conn, err := amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")
	return conn, err
}

// 发送端函数
func Publist(exchange string, queueName string, body string) error {
	conn, err := Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	// 创建通道 channel
	channel, err := conn.Channel()
	if err != nil {
		return err
	}
	defer channel.Close()

	// 创建队列 queue
	q, err := channel.QueueDeclare(
		queueName, // 队列名
		true,      // 是否持久化
		false,     // 自动删除
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}

	//发送消息
	err = channel.Publish(exchange, q.Name, false, false, amqp.Publishing{
		DeliveryMode: amqp.Persistent, //消息的持久化
		ContentType:  "text/plain",    //类型
		Body:         []byte(body),
	})
	return err
}

type Callback func(msg string)

// 接收者方法
func Consumer(exchange string, queueName string, callback Callback) {
	conn, err := Connect()
	if err != nil {
		fmt.Println(err)
		return
	}
	defer conn.Close()

	// 创建通道 channel
	channel, err := conn.Channel()
	if err != nil {
		fmt.Println(err)
		return
	}
	defer channel.Close()

	// 创建队列 queue
	q, err := channel.QueueDeclare(
		queueName, // 队列名
		true,      // 持久化
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 从队列获取数据  //第三参数:自动应答, 应答后自动删除消息
	msgs, err := channel.Consume(q.Name, "", false, false, false, false, nil)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 启动channel
	forever := make(chan bool)

	// 这是一个守护进程,不能退出 , 所以启动协程
	go func() {
		for d := range msgs {
			s := BytesToString(&(d.Body))
			callback(*s)
			d.Ack(false) // 手动应答, 应答后自动删除消息
		}
	}()
	fmt.Println("waiting for message")