Golang学习条记_RabbitMQ的原理架构和使用
RabbitMQ 简介[*]实现了高级消息队列协议（Advanced Message Queuing Protcol）AMQP
[*]消息队列中央件的作用（Redis实现MQ里面有写过，这里简朴带过）
[*]解耦
[*]削峰
[*]异步处理
[*]缓存
[*]消息通信
[*]提高扩展性
RabbitMQ 架构明确
[*]binding(绑定)：互换机将消息路由给Queue所遵循的规则，可以界说一个路由键，用于互换机筛选特定的Queue
[*]Routing_Key(路由键)：Producer 和 Consumer 协商一致的 key 策略。主要在互换机的 direct(直连)和 topic(主题) 模式下使用，fanout(广播)模式下不使用Routing_Key
[*]Exchange(互换机)：主要功能是分发消息给特定的Queue，只负责转发，不具备存储消息的功能。Exchange有以下四种模式：
[*]direct(直连模式)，根据携带的Routing_Key来筛选特定的Queue进行消息投递。是RabbitMQ的默认类型，可以不指定Routing_Key，在创建时会默认生成与Queue重名。
[*]hander(头模式)，使用场景不多，消息路由涉及多个属性的时候，互换机使用多属性来取代Routing_key创建路由规则，还可以界说匹配单词的个数，例如any为有一个单词满意条件就匹配乐成。all为所有单词都满意条件才匹配乐成。
[*]fanout(广播模式)，不看Routing_Key。只根据Exchange和Queue的binding情况来分发信息。所有与之binding的queue都将接收到同一条消息。
[*]topic(主题模式)，相当于模糊查询。topic的routing_key是使用 . 来进行隔断的。有两种匹配方法：
[*]" * " 匹配一个单词，例子如下
[*]" # " 匹配0个~多个单词，例子如下
rabbitMQ.* == rabbitMQ.topic != rabbitMQ.topic.topic
rabbitMQ.# == rabbit.topic == rabbit.topic.topic
[*]Queue（消息队列的存储数据布局）：
[*]存储方式：
[*]持久化，在Server当地硬盘存储一份
[*]临时队列，重启后丢失数据
[*]自动删除，不存在用户连接则删除queue
[*]队列对ACK请求的不同情况
[*]consumer 接收并 ack，queue 删除数据并向 consumer 发送新消息
[*]consumer 接收但是未 ack 就断开了连接，queue 会认为消息并未传送乐成，consumer 再次连接时会重新发送消息
[*]如果consumer 接收消息乐成 ，但是忘记 ack 则 queue 不会重复发送消息
[*]如果 consumer 拒收消息，则 queue 会向另外满意条件的 consumer 继承发送这条消息
RabbitMQ 工作流程
Producer方向
[*]Producer 与 RabbitMQ Broker 创建连接，开启一个信道 channel
[*]声明互换机并设置属性（互换机类型、持久化等）
[*]声明Queue并设置属性（持久化，自动删除等）
[*]通过Routing_key来binding互换机和Queue
[*]发送信息给互换，互换机根据Routing_key来确认投递的queue
[*]查找乐成后将消息存到queue
[*]查找失败将消息丢弃或抛回给生产者
[*]关闭channel
Consumer方向
[*]与 queue 创建连接，开启channel
[*]向queue请求队列中的msg
[*]等待queue回应，开始接收消息
[*]消息处理完成后 返回回调确认ack
[*]queue 将确认的消息从队列中删除
[*]关闭channel
RabbitMQ的两种摆设方式
Meta Data ： 元数据（形貌数据的数据）
[*]vhost meta data : 为Queue、Exchange、Binding提供命名空间级别的隔离
[*]exchange meta data：记录路由的名称类型和属性
[*]binding mate data：映射 routing_key和queue之间的绑定关系
[*]queue mate data：表队列名称和属性
普通模式
对于该模式的两个节点，消息只会存在其中一个节点，另一个节点只保存mate data，当consumer 连接节点2访问节点1的数据信息时，消息会在两个节点中传递。
该模式下p和c应尽量连接每个节点，这样起到线性拓展的作用。
但存在一个问题，如果节点上另有未消费的消息，但是节点挂了。如果节点设置了持久化，则需要在节点重启的时候消息才会规复。如果未设置持久化，则消息会丢失。
镜像模式
消息存在多个节点中，消息会在节点与节点之间同步，可实现高可用（当一个节点挂了，另一个节点可以接替其位置，继承工作）但会低落性能，由于大量消息进入和同步，会占用大量带宽，但是为了保证高可靠性需要弃取。
面试题
[*]Q：如何保证消息不被重复消费？
[*]A：MQ通过确认机制ACK，进行确认。确认后消息从queue中删除，保证消息不被重复消费的。如果由于网络原因ack没有乐成发出，导致消息重新投递。可以使用全局唯一消息id来制止。
[*]消息发送者发送消息时携带一个全局唯一的消息id
[*]消费者监听到消息后，根据id在redis或者db中查询是否存在消费记录
[*]如果没有消费就正常消费，消费完毕后，写入redis或者db
[*]如果消息消费过则直接丢弃
[*]Q：如何保证消息的消费次序？
[*]A：RabbitMQ中存在一个设置，叫独占队列。即在同一时间只有一个消费者会消费消息。从而制止了异步操作，保证消费次序。或者一个Producer对一个Consumer
[*]Q：如何保证数据一致性？
[*]A：由于MQ的使用场景多为分布式系统，以是一般不寻求强一致性。而保证终极一致性就可以。
[*]而保证数据终极一致性，可以采用消息赔偿机制。即消息在消费者处理完之后调用生产者的API修改数据状态。如未调用API则判断为消息处理失败或出错。此时间隔一段时间后重新投递消息进行再次操作。
[*]消费者收到消息，处理完毕后，发送一条响应消息给生产者也是消息赔偿机制，本意是确认消费者乐成消费消息。ACK也是处理方法
RabbitMQ的使用（Golang使用amqp包）
代码部分参考 upup小亮的博客
代码只是简朴的操作，主要是认识流程。对于如何创建Queue和绑定Exchange之类的操作有个相识。
Simple（简朴收发模式，只有一个Queue）
Simple运行机制与WorkQueue相似，只是一个Consumer与多个Consumer的区别。多个Consumer之间存在竞争关系，以是工作队列是创建多个Consumer，多个竞争只有一个可以获取消息消费。消费乐成后ack消息删除。
演示代码放到一起了：
WorkQueue 工作队列
生产者
// simple and work queue
func main2() {
// 连接到 rabbitMQ
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("无法创建连接:%s", err)
return
}
// 默认关闭
defer conn.Close()
// 创建通道Channel
ch, err := conn.Channel()
if err != nil {
log.Fatalf("无法创建channel:%s", err)
return
}
// 通道关闭
defer ch.Close()
// 创建存储队列
queue, err := ch.QueueDeclare(
"hello", // 队列名称
false, // 持久化设置，可以为true根据需求选择
false, // 自动删除，没有用户连接删除queue一般不选用
false, //独占
false, //等待服务器确认
nil) //参数
if err != nil {
fmt.Println(err)
log.Fatalf("无法声明队列:%s", err)
return
}
var body string
// 发送信息
for i := 0; i < 10; i++ {
fmt.Println(i)
body = "Hello RabbitMQ" + string(i)
err = ch.Publish(
"",
queue.Name,
false, // 必须发送到消息队列
false, // 不等待服务器确认
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
if err != nil {
log.Fatalf("消息生产失败:%s", err)
continue
}
}
}
消费者
// create conn
// 如果同时运行两个这样的consumer代码，就是工作队列。只有一个consumer就是simple
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("无法创建连接:%s", err)
return
}
defer conn.Close()
// create channel
ch, err := conn.Channel()
if err != nil {
log.Fatalf("无法创建channel:%s", err)
return
}
defer ch.Close()
// create queue
queue, err := ch.QueueDeclare(
"hello",
false,
false,
false,
false,
nil)
if err != nil {
log.Fatalf("无法创建queue:%s", err)
return
}
// 消费信息
msgs, err := ch.Consume(
queue.Name,
"",
true,
false,
false,
false,
nil)
if err != nil {
log.Fatalf("无法消费信息:%s", err)
return
}
for msg := range msgs {
log.Println(string(msg.Body))
}
return
pub/sub 发布订阅模式
发布订阅模式可以创建两个Queue，绑定到同一个Exchange中
生产者这边只需要跟互换机对接，而互换机类型为fanout：
func main() {
// 连接到 rabbitMQ
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("无法创建连接:%s", err)
}
// 默认关闭
defer conn.Close()
// 创建通道Channel
ch, err := conn.Channel()
if err != nil {
log.Fatalf("无法创建channel:%s", err)
}
defer ch.Close()
// create exchange
ex := ch.ExchangeDeclare(
"exchange1", // 交换机名称
"fanout", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待服务器响应
nil, // 其他属性
)
fmt.Println(ex)
body := "Hello RabbitMQ for Pub/Sub"
err = ch.Publish(
"exchange1",
"", // routing key 可以为空，因为fanout不看routing key
false,
false,
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
})
if err != nil {
log.Fatalf("err %s:", err)
}
log.Println(body)
}
消费者：创建互换机，类型为fanout，创建队列，绑定互换机（创建多个consumer绑定同一个queue和同一个互换机。这样发送一个消息，所有的consumer都能收到。== 发布订阅模型）
// Pub/Sub
// Create conn
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil{
log.Fatalf(err)
}
defer conn.Close()
// channel create
ch, err := conn.Channel()
if err != nil{
log.Fatalf(err)
}
defer ch.Close()
// exchange create
ex := ch.ExchangeDeclare(
"exchange1",
"fanout",
true,
false,
false,
false,
nil)
fmt.Println(ex)
// queue create
queue, err := ch.QueueDeclare(
"hello",
false,
false,
false,
false,
nil)
if err != nil{
log.Fatalf(err)
}
err = ch.QueueBind(
queue.Name,
"",
"exchange1",
false,
nil)
if err != nil{
log.Fatalf(err)
}
msgs, err := ch.Consume(
queue.Name,
"",
true,
false,
false,
false,
nil)
if err != nil{
log.Fatalf(err)
}
go func() {
for d := range msgs {
log.Printf("Received a message: %s", d.Body)
}
}()
log.Printf("Waiting for messages. To exit press CTRL+C")
<-make(chan struct{}) // 阻塞主goroutine
}
Routing 模式（对特定的队列投递消息）
生产者
func main() {
// 连接到 rabbitMQ
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
log.Fatalf("无法创建连接:%s", err)
}
// 默认关闭
defer conn.Close()
// 创建通道Channel
ch, err := conn.Channel()
if err != nil {
log.Fatalf("无法创建channel:%s", err)
}
defer ch.Close()
// create exchange
ex := ch.ExchangeDeclare(
"exchange1", // 交换机名称
"direct", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待服务器响应
nil, // 其他属性
)
fmt.Println(ex)
body := "Hello RabbitMQ for direct routing"
// 发布消息到交换机，并指定路由键
err = ch.Publish(
"logs_direct", // 交换机名称
"routing_key", // 路由键
false, // 是否等待服务器响应
false, // 是否立即将消息写入磁盘
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(body),
},
)
if err != nil{
log.Fatalf("无法创建send msg:%s", err)
}
log.Printf("Sent message: %s", message)
消费者
func main() {
// 连接到RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil{
log.Fatalf("无法创建send msg:%s", err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil{
log.Fatalf("无法创建send msg:%s", err)
}
defer ch.Close()
// 声明一个交换机
err = ch.ExchangeDeclare(
"logs_direct", // 交换机名称
"direct", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf("无法创建send msg:%s", err)
}
// 声明一个临时队列
q, err := ch.QueueDeclare(
"", // 队列名称，留空表示由RabbitMQ自动生成,因为定义了key所以队列名可以是随意的，毕竟是依靠key来进行匹配的
false, // 是否持久化
false, // 是否自动删除（当没有任何消费者连接时）
true,// 是否排他队列（仅限于当前连接）
false, // 是否等待服务器响应
nil, // 其他属性
)
// 将队列绑定到交换机上，并指定要接收的路由键
err = ch.QueueBind(
q.Name, // 队列名称
"routing_key", // 路由键
"logs_direct", // 交换机名称
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf("无法创建send msg:%s", err)
}
// 订阅消息
msgs, err := ch.Consume(
q.Name, // 队列名称
"", // 消费者标识符，留空表示由RabbitMQ自动生成
true, // 是否自动应答
false,// 是否独占模式（仅限于当前连接）
false,// 是否等待服务器响应
false,// 其他属性
nil, // 其他属性
)
failOnError(err, "Failed to register a consumer")
// 接收消息的goroutine
go func() {
for d := range msgs {
log.Printf("Received a message: %s", d.Body)
}
}()
log.Printf("Waiting for messages. To exit press CTRL+C")
<-make(chan struct{}) // 阻塞主goroutine
topic
func main() {
// 连接到RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil{
log.Fatalf(err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil{
log.Fatalf(err)
}
defer ch.Close()
// 声明一个交换机
err = ch.ExchangeDeclare(
"logs_topic", // 交换机名称
"topic", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf(err)
}
// 定义要发送的消息的路由键和内容
routingKey := "example.key.das"
message := "Hello, RabbitMQ!"
// 发布消息到交换机，并指定路由键
err = ch.Publish(
"logs_topic", // 交换机名称
routingKey, // 路由键
false, // 是否等待服务器响应
false, // 是否立即发送
amqp.Publishing{
ContentType: "text/plain",
Body: []byte(message),
},
)
if err != nil{
log.Fatalf(err)
}
log.Printf("Sent message: %s", message)
}
消费者
func main() {
// 连接到RabbitMQ服务器
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil{
log.Fatalf(err)
}
defer conn.Close()
// 创建一个通道
ch, err := conn.Channel()
if err != nil{
log.Fatalf(err)
}
defer ch.Close()
// 声明一个交换机
err = ch.ExchangeDeclare(
"logs_topic", // 交换机名称
"topic", // 交换机类型
true, // 是否持久化
false, // 是否自动删除
false, // 是否内部使用
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf(err)
}
// 声明一个临时队列
q, err := ch.QueueDeclare(
"", // 队列名称，留空表示由RabbitMQ自动生成
false, // 是否持久化
false, // 是否自动删除（当没有任何消费者连接时）
true,// 是否排他队列（仅限于当前连接）
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf(err)
}
// 将队列绑定到交换机上，并指定要接收的路由键
err = ch.QueueBind(
q.Name, // 队列名称
"example.#",// 路由键，可以使用通配符*匹配一个单词
"logs_topic", // 交换机名称
false, // 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf(err)
}
// 创建一个消费者通道
msgs, err := ch.Consume(
q.Name, // 队列名称
"", // 消费者标识符，留空表示由RabbitMQ自动生成
true, // 是否自动应答
false,// 是否排他消费者
false,// 是否阻塞
false,// 是否等待服务器响应
nil, // 其他属性
)
if err != nil{
log.Fatalf(err)
}
// 接收和处理消息
forever := make(chan bool)
go func() {
for d := range msgs {
log.Printf("Received a message: %s", d.Body)
}
}()
log.Printf("Waiting for messages...")
// 阻塞
<-forever
}
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]