RabbitMq如何做到消息的可靠性投递
作者:Onemorelight95 发布时间:2024-04-23 09:36:57
如何保证消息不丢失
在使用RabbitMQ的时候,我们需要保证消息不能丢失,消息从生产者生产出来一直到消费者消费成功,这条链路是这样的:
消息的可靠投递分为了两大内容:发送端的确认(p->broker和exchange->queue)和消费端的确认(queue->c)。
发送端的确认
Rabbit提供了两种方式来保证发送端的消息可靠性投递:confirm 确认模式
和return 退回模式。
confirm 确认模式:消息从 producer 到达 exchange 则会给 producer 发送一个应答,我们需要开启confirm模式,才能接收到这条应答。开启方式是将Channel.Confirm(noWait bool)
参数设置为false
,表示同意发送者将当前channel信道设置为confirm模式。
return 退回模式:消息从 exchange–>queue 投递失败,会将消息退回给producer。
消费端的确认
消息从Queue发送到消费端之后,消费端会发送一个确认消息:Consumer Ack
,有两种确认方式:自动确认和手动确认。
在编码中,关于消息的确认方式,我们需要在消费者端调用Consumer
函数时,设置第三个参数:autoAck
是false还是true(false表示手动,true表示自动)。
自动确认是指,当消息一旦被Consumer接收到,则自动确认收到,并将相应 message 从 RabbitMQ 的消息缓存中移除。
但是在实际业务处理中,很可能消息接收到,业务处理出现异常,那么该消息就会丢失。如果设置了手动确认方式,则需要在业务处理成功后,调用ch.Ack(false)
,手动签收,如果出现异常,则调用d.Reject(true)
让其自动重新发送消息。
Go 实现
安装操作库
安装API库
Go可以使用streadway/amqp
库来操作rabbit,使用以下命令来安装:
go get github.com/streadway/amqp
封装rabbitmq
接下来我们对streadway/amqp
库的内容进行一个二次封装,封装为一个rabbitmq.go
文件:
package rabbitmq
import (
"encoding/json"
"github.com/streadway/amqp"
"log"
)
// RabbitMQ RabbitMQ结构
type RabbitMQ struct {
channel *amqp.Channel
Name string
exchange string
}
// Connect 连接服务器
func Connect(s string) *RabbitMQ {
//连接rabbitmq
conn, e := amqp.Dial(s)
failOnError(e, "连接Rabbitmq服务器失败!")
ch, e := conn.Channel()
failOnError(e, "无法打开频道!")
mq := new(RabbitMQ)
mq.channel = ch
return mq
}
// New 初始化消息队列
//第一个参数:rabbitmq服务器的链接,第二个参数:队列名字
func New(s string, name string) *RabbitMQ {
//连接rabbitmq
conn, e := amqp.Dial(s)
failOnError(e, "连接Rabbitmq服务器失败!")
ch, e := conn.Channel()
failOnError(e, "无法打开频道!")
q, e := ch.QueueDeclare(
name, //队列名
false, //是否开启持久化
true, //不使用时删除
false, //排他
false, //不等待
nil, //参数
)
failOnError(e, "初始化消息队列失败!")
mq := new(RabbitMQ)
mq.channel = ch
mq.Name = q.Name
return mq
}
// QueueDeclare 声明queue
func (q *RabbitMQ) QueueDeclare(queue string) {
_, e := q.channel.QueueDeclare(queue, false, true, false, false, nil)
failOnError(e, "声明queue失败!")
}
// QueueDelete 删除queue
func (q *RabbitMQ) QueueDelete(queue string) {
_, e := q.channel.QueueDelete(queue, false, true, false)
failOnError(e, "删除queue失败!")
}
// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
e := q.channel.Qos(1, 0, false)
failOnError(e, "无法设置QoS")
}
// NewExchange 初始化交换机
//第一个参数:rabbitmq服务器的链接,第二个参数:交换机名字,第三个参数:交换机类型
func NewExchange(s string, name string, typename string) {
//连接rabbitmq
conn, e := amqp.Dial(s)
failOnError(e, "连接Rabbitmq服务器失败!")
ch, e := conn.Channel()
failOnError(e, "无法打开频道!")
e = ch.ExchangeDeclare(
name, // name
typename, // type
true, // durable
false, // auto-deleted
false, // internal
false, // no-wait
nil, // arguments
)
failOnError(e, "初始化交换机失败!")
}
// ExchangeDelete 删除交换机
func (q *RabbitMQ) ExchangeDelete(exchange string) {
e := q.channel.ExchangeDelete(exchange, false, true)
failOnError(e, "删除交换机失败!")
}
// Bind 绑定消息队列到exchange
func (q *RabbitMQ) Bind(exchange string, key string) {
e := q.channel.QueueBind(
q.Name,
key,
exchange,
false,
nil,
)
failOnError(e, "绑定队列失败!")
q.exchange = exchange
}
// Send 向消息队列发送消息
//Send方法可以往某个消息队列发送消息
func (q *RabbitMQ) Send(queue string, body interface{}) {
str, e := json.Marshal(body)
failOnError(e, "消息序列化失败!")
e = q.channel.Publish(
"", //交换
queue, //路由键
false, //必填
false, //立即
amqp.Publishing{
ReplyTo: q.Name,
Body: []byte(str),
})
msg := "向队列:" + q.Name + "发送消息失败!"
failOnError(e, msg)
}
// Publish 向exchange发送消息
//Publish方法可以往某个exchange发送消息
func (q *RabbitMQ) Publish(exchange string, body interface{}, key string) {
str, e := json.Marshal(body)
failOnError(e, "消息序列化失败!")
e = q.channel.Publish(
exchange,
key,
false,
false,
amqp.Publishing{ReplyTo: q.Name,
Body: []byte(str)},
)
failOnError(e, "向交换机发送消息失败!")
}
// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
c, e := q.channel.Consume(
q.Name, //指定从哪个队列中接收消息
"",
true,
false,
false,
false,
nil,
)
failOnError(e, "接收消息失败!")
return c
}
// Close 关闭队列连接
func (q *RabbitMQ) Close() {
q.channel.Close()
}
//错误处理函数
func failOnError(err error, msg string) {
if err != nil {
log.Fatalf("%s: %s", msg, err)
}
}
发送端的确认
首先初始化消息队列的时候,我们要开启confirm模式
,才能接收到这条应答。开启方式是将Channel.Confirm(noWait bool)
参数设置为false
,表示同意发送者将当前channel信道设置为confirm模式。
func New(s string, name string) *RabbitMQ {
conn, e := amqp.Dial(s)
failOnError(e, "连接Rabbitmq服务器失败!")
ch, e := conn.Channel()
failOnError(e, "无法打开频道!")
q, e := ch.QueueDeclare(
name, //队列名
false, //是否开启持久化
true, //不使用时删除
false, //排他
false, //不等待
nil, //参数
)
failOnError(e, "初始化消息队列失败!")
mq := new(RabbitMQ)
mq.channel = ch
mq.Name = q.Name
// 设置为confirm模式
mq.channel.Confirm(false)
return mq
}
然后在封装库中创建一个函数handleConfirm()
用于接收来自Borker的回复:
func (q *RabbitMQ) ConfirmFromBroker(ch chan amqp.Confirmation) chan amqp.Confirmation {
return q.channel.NotifyPublish(ch)
}
生产者
生产者端在向Broker发送消息的时候,我们使用一个无缓冲的通道来接收来自Broker的回复,然后创建一个协程监听这个无缓冲通道。
func main() {
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
// 指定为topic类型
rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange1", "fanout")
confirm := producer.ConfirmFromBroker(make(chan amqp.Confirmation))
go handleConfirm(confirm)
var i int
for {
time.Sleep(time.Second)
producer.Publish("exchange1", "fanout message: "+strconv.Itoa(i), "")
i++
}
}
func handleConfirm(confirm <-chan amqp.Confirmation) {
for {
select {
case message := <-confirm:
fmt.Println("接收到来自Broker的回复:", message)
}
}
}
运行结果:
接收到来自Broker的回复: {1 true}
接收到来自Broker的回复: {2 true}
接收到来自Broker的回复: {3 true}
接收到来自Broker的回复: {4 true}
接收到来自Broker的回复: {5 true}
消费端的确认
首先将Consume
函数的第三个参数autoAck
参数标记为false:
// Consume 接收某个消息队列的消息
func (q *RabbitMQ) Consume() <-chan amqp.Delivery {
c, e := q.channel.Consume(
q.Name,
"",
false, // 不自动确认消息
false,
false,
false,
nil,
)
failOnError(e, "接收消息失败!")
return c
}
在消费者端我们采用公平派遣模式,即队列发送消息给消费者的时候,不再采用轮询机制,而是一个消费者消费完消息之后,会调用Ack(false)
函数向队列发送一个回复,队列每次会将消息优先发送给消费完消息的消费者(回复过)。
消费端限流:
实现公平派遣模式我们需要设置消费者端一次只能消费一条消息,之前我们已经进行了封装,直接在消费者端调用即可:
// Qos 配置queue参数
func (q *RabbitMQ) Qos() {
e := q.channel.Qos(1, 0, false)
failOnError(e, "无法设置QoS")
}
生产者
func main() {
producer := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue")
// 指定为direct类型
rabbitmq.NewExchange("amqp://guest:guest@35.76.111.125:5672/", "exchange", "direct")
i := 0
for {
time.Sleep(time.Second)
producer.Publish("exchange", "routing message: "+strconv.Itoa(i), "key1")
i = i + 1
}
}
消费者1
消费者2在消费第三条消息的时候,假设发生了错误,我们调用d.Reject(true)
函数让队列重新发送消息。
func main() {
//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
consumer1 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
// 指定一次只消费一条消息,直到消费完才重新接收
consumer1.Qos()
// 队列绑定到exchange
consumer1.Bind("exchange", "key1")
//接收消息
msgs := consumer1.Consume()
go func() {
var i int
for d := range msgs {
time.Sleep(time.Second * 1)
log.Printf("Consumer1 received a message: %s", d.Body)
// 假设消费第三条消息的时候出现了错误,我们就调用d.Reject(true),队列会重新发送消息给消费者
if i == 2 {
d.Reject(true)
} else {
// 消息消费成功之后就回复
d.Ack(false)
}
i++
}
}()
select {}
}
消费者2
func main() {
//第一个参数指定rabbitmq服务器的链接,第二个参数指定创建队列的名字
consumer2 := rabbitmq.New("amqp://guest:guest@35.76.111.125:5672/", "queue1")
// 指定一次只消费一条消息,直到消费完才重新接收
consumer2.Qos()
// 队列绑定到exchange
consumer2.Bind("exchange", "key1")
//接收消息
msgs := consumer2.Consume()
go func() {
for d := range msgs {
time.Sleep(time.Second * 5)
log.Printf("Consumer2 received a message: %s", d.Body)
// 消息消费成功之后就回复
d.Ack(false)
}
}()
select {}
}
运行结果:
# 消费者1
2022/11/06 19:55:08 Consumer1 received a message: "routing message: 0"
2022/11/06 19:55:10 Consumer1 received a message: "routing message: 2"
2022/11/06 19:55:11 Consumer1 received a message: "routing message: 3"
2022/11/06 19:55:12 Consumer1 received a message: "routing message: 3"
2022/11/06 19:55:13 Consumer1 received a message: "routing message: 4"
2022/11/06 19:55:14 Consumer1 received a message: "routing message: 6"# 消费者2
2022/11/06 19:55:13 Consumer2 received a message: "routing message: 1"
来源:https://blog.csdn.net/qq_49723651/article/details/127717187
猜你喜欢
- 一、Cookie的使用1、什么是Cookie我们都知道,HTTP 协议是无状态的,也就是说,在一次请求响应结束后,服务端不会保留任何对方状态
- 本文介绍了python OpenCV学习笔记直方图反向投影的实现,分享给大家,具体如下:官方文档 – https://docs.opencv
- 1. resize函数说明 OpenCV提供了resize函数来改变图像的大小,函数原型如下:void resiz
- 在django操作数据库的时候如何找出内容不为空的数据呢?from django.db.models import Qclass Index
- 这篇文章主要介绍了如何通过Django使用本地css/js文件,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,
- 在图书馆的检索系统中,关于图书的信息里面有一个是图书相关借阅关系图。跟这个社交网络图是一样的,反映了不同对象间的关联性。利用python画社
- 看到网上有一篇文章,是介绍如何做网站测试的.从一开始的链接测试,页面内容测试,到浏览器兼容性测试,负载压力测试,一直到最后的安全性测试,甚至
- 使用SQL语句从电脑导入图片到数据库的方法,具体代码如下所示:--创建图片表CREATE TABLE W_PIC( ID INT
- 一、Selects检索表中的所有行$users = DB::table('users')->get();foreach
- 项目测试对于一个项目的重要性,大家应该都知道吧,写python的朋友,应该都写过自动化测试脚本。最近正好负责公司项目中的api测试,下面写了
- python脚本性能分析首先使用cd进入需要测试的脚本文件对应的目录,然后再使用如下代码完成对脚本的性能测试。# enter the dir
- 本文实例讲述了python通过pil模块将raw图片转换成png图片的方法。分享给大家供大家参考。具体分析如下:python通过pil模块将
- MySQL 默认有个root用户,但是这个用户权限太大,一般只在管理数据库时候才用。如果在项目中要连接 MySQL 数据库,则建议新建一个权
- 一、表结构TABLE personidname1你2你(一个空格)3你(二个空格)二、查询与结果select * from person w
- 目录循环加判断retrying我们在程序开发中,经常会需要请求一些外部的接口资源,而且我们不能保证每次请求一定会成功,所以这些涉及到网络请求
- stringObject.substring(start,stop) 用于提取字符串中介于两个指定下标之间的字符。start必需。一个非负的
- 信号(signals)Flask信号(signals, or event hooking)允许特定的发送端通知订阅者发生了什么(既然知道发生
- 提要:系统自带的mysql默认字符集不是gbk,因此给数据库的推广应用以及中文程序的开发带来极大的不便,在没完没了的GBK和UTF8的转换过
- Oracle游标分为显示游标和隐式游标。显示游标(Explicit Cursor):在PL/SQL程序中定义的、用于查询的游标称作显示游标。
- 前言Python 是每个程序员都喜欢的语言,因为它易于编码和易于阅读的语法。但是,你知道 python 有一些很酷的技巧可以用来让事情变得更