beego - rabbitMq持久化 - 简单模式工作模式

服务重启数据丢失

必须发送端和接收端都设置持久化才可以

1. 发送端创建队列时 , 第二参数(是否持久化)改成true

2. 发送端发送消息时在amqp.Publishing中添加

DeliveryMode: amqp.Persistent

3. 接收端创建队列时 , 第二参数(是否持久化)改成true

接收者服务挂掉数据丢失

1. 从队列中获取数据channel.Consumer() ,

第三参数:自动应答, 应答后自动删除消息 , 修改为false

2. 回调函数后 , 添加Ack()

	go func() {
		for d := range msgs {
			s := BytesToString(&(d.Body))
			callback(*s)
			d.Ack(false) // 手动应答, 应答后自动删除消息
		}
	}()

原封装代码:

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, // 队列名
		false,     // 是否持久化
		false,     // 自动删除
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}

	//发送消息
	err = channel.Publish(exchange, q.Name, false, false, amqp.Publishing{
		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,
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		fmt.Println(err)
		return
	}

	// 从队列获取数据
	msgs, err := channel.Consume(q.Name, "", true, 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)
		}
	}()
	fmt.Println("waiting for message")
	<-forever
}

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

修改后:

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
}