beego RabbitMQ 简单模式 工作模式

MQ服务  

package mq

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

// 连接
func Connect() (*amqp.Connection, error) {
	conn, err := amqp.Dial("amqp://guest:[email protected]: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")
	<-forever
}

// 字节转字符串
func BytesToString(b *[]byte) *string {
	s := bytes.NewBuffer(*b)
	r := s.String()
	return &r
}

发送者 

package controllers

import (
	"github.com/astaxie/beego"
	"strconv"
	"time"
	"youku/services/mq"
)

type MqController struct {
	beego.Controller
}

// 简单模式和 work工作模式 push
// @router /mq/push [*]
func (this *MqController) MqPush() {
	go func() {
		count := 0
		for {
			mq.Publist("", "youku_demo", "hello"+strconv.Itoa(count))
			count++
			time.Sleep(1 * time.Second)
		}
	}()
	this.Ctx.WriteString("发送")
}

接受者

package main

import (
	"fmt"
	"youku/services/mq"
)

func main() {
	mq.Consumer("", "youku_demo", callback)
}

func callback(s string) {
	fmt.Printf("msg is :%s\n", s)
}