rabbitmq原理和应用

0.1、索引

https://blog.waterflow.link/articles/1663772504649

RabbitMQ 是一个轻量级且易于部署的消息队列。它支持开箱即用的多种消息传递协议。我们将使用 AMQP（高级消息队列协议）

1、概念

既然是消息队列，顾名思义，肯定会有生产者生产消息，消费者消费消息，还会有队列用来保存消息，等等。

我们先来看下这些概念：

• Producer: 将消息推送到 rabbitmq 交换机的应用
• Consumer: 从队列读取消息并处理他们的应用
• Exchange: 交换机负责在 Binding 和 Routing key 的帮助下，将消息路由到不同的队列。从上图可以看出 rabbitmq 有多种类型的交换机
• Binding: Binding 是队列和交换机之间的链接
• Routing key: 交换机用来决定如何将消息路由到队列的键。可以看做是消息的地址
• Queue: 存储消息的缓冲区
• **Connection:**生产者到 Broker（rabbitmq 服务），消费者到 Broker 的连接
• **Channel：**为了复用一个连接，一个 connection 下可以有多个 channel，可以把 connection 理解成电线，channel 就是电线里面的铜丝。

消息传递的完整流程是这样的：

1. 生产者初始化一个到 rabbitmq 服务的连接
2. 获取连接的管道，通过管道声明一个交换机
3. 通过管道声明一个队列，通过绑定的路由键将队列和交换机绑定（发送消息的时候声明一个队列并绑定交换机，消息会进到队列里。如果不声明也可以放到消费者去声明队列和绑定交换机。需要注意的是生产者没有声明队列的话，此时已经生产多条消息，然后去开启消费者消费，是不会消费到之前的消息的）
4. 通过管道发送消息到指定的交换机
5. 消费者初始化一个到 rabbitmq 服务的连接
6. 获取连接的管道，通过管道声明一个队列
7. 通过绑定的路由键将队列和交换机绑定
8. 从队列中消费消息

交换机类型：

1. **direct：**直接指定到某个队列
2. **topic：**发布订阅模式，一个交换机可以对应多个队列，通过路由规则匹配
3. **fanout：**顾名思义，无脑广播模式

2、示例

生产者：

package main

import (
	"fmt"
	"time"

	"github.com/streadway/amqp"
)

var (
	conn    *amqp.Connection
	channel *amqp.Channel
	queue   amqp.Queue
	mymsg   = "Hello HaiCoder"
	err     error

	confirms chan amqp.Confirmation
)

func main() {
	// 建立连接
	conn, err = amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")
	if err != nil {
		fmt.Println(err)
		return
	}

	defer conn.Close()

	// 创建channel
	if channel, err = conn.Channel(); err != nil {
		fmt.Println(err)
		return
	}

	// 声明交换机
	err = channel.ExchangeDeclare("liutest", amqp.ExchangeDirect, false, false, false, false, nil)
	if err != nil {
		fmt.Println("ExchangeDeclare Err =", err)
		return
	}

	// 创建队列
	if queue, err = channel.QueueDeclare("liutest", false, false, false, false, nil); err != nil {
		fmt.Println("QueueDeclare Err =", err)
		return
	}

	// 队列和交换机绑定
	err = channel.QueueBind(queue.Name, "queueroutekey", "liutest", false, nil)
	if err != nil {
		fmt.Println("QueueBind Err =", err)
		return
	}

	channel.Confirm(false)
	confirms = channel.NotifyPublish(make(chan amqp.Confirmation, 1))
	//发送数据
	go func() {
		for {
			if err = channel.Publish("liutest", "queueroutekey", false, false, amqp.Publishing{
				ContentType: "text/plain",
				Body:        []byte(mymsg),
			}); err != nil {
				fmt.Println("Publish Err =", err)
				return
			}
			fmt.Println("Send msg ok, msg =", mymsg)
			time.Sleep(time.Second * 5)
		}
	}()

	go func() {
		for confirm := range confirms {
			if confirm.Ack {
				fmt.Printf("confirmed delivery with delivery tag: %d \n", confirm.DeliveryTag)
			} else {
				fmt.Printf("confirmed delivery of delivery tag: %d \n", confirm.DeliveryTag)
			}
		}
	}()

	select {}

}

消费者：

package main

import (
	"fmt"

	"github.com/streadway/amqp"
)

var (
	conn    *amqp.Connection
	channel *amqp.Channel
	queue   amqp.Queue
	err     error
	msgs    <-chan amqp.Delivery
)

func main() {
	// 建立连接
	conn, err = amqp.Dial("amqp://guest:guest@127.0.0.1:5672/")
	if err != nil {
		fmt.Println(err)
		return
	}

	defer conn.Close()

	// 创建channel
	if channel, err = conn.Channel(); err != nil {
		fmt.Println(err)
		return
	}

	// 创建队列
	if queue, err = channel.QueueDeclare("liutest", false, false, false, false, nil); err != nil {
		fmt.Println("QueueDeclare Err =", err)
		return
	}

	err = channel.QueueBind("liutest", "queueroutekey", "liutest", false, nil)
	if err != nil {
		fmt.Println("QueueBind Err =", err)
		return
	}
	//读取数据
	if msgs, err = channel.Consume(queue.Name, "", false, false, false, false, nil); err != nil {
		fmt.Println("Consume Err =", err)
		return
	}
	go func() {
		for msg := range msgs {
			fmt.Println("Receive Msg =", string(msg.Body))
			msg.Ack(false)
		}
	}()

	select {}

}

3、消息可靠性

生产者可靠性

// 将通道设置为确认模式
func (ch *Channel) Confirm(noWait bool) error {
	if err := ch.call(
		&confirmSelect{Nowait: noWait},
		&confirmSelectOk{},
	); err != nil {
		return err
	}

	ch.confirmM.Lock()
	ch.confirming = true
	ch.confirmM.Unlock()

	return nil
}

// 用于接受服务端的确认响应
func (ch *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation {
	ch.notifyM.Lock()
	defer ch.notifyM.Unlock()

	if ch.noNotify {
		close(confirm)
	} else {
		ch.confirms.Listen(confirm)
	}

	return confirm

}

Confirm 将此通道置为确认模式，以便生产者可以确保服务端已成功接收所有消息。进入该模式后，服务端将发送一个 basic.ack 或 basic.nack 消息，其中 deliver tag 设置为一个基于 1 的增量索引（用来标识消息的唯一性），对应于该方法返回后收到的每次 ack。

在 Channel.NotifyPublish 上监听以响应 ack。如果未调用 Channel.NotifyPublish，则 ack 将被忽略。

ack 的顺序不受投递消息顺序的约束。

Ack 和 Nack 确认将在未来的某个时间到达。

在通知任何 Channel.NotifyReturn 侦听器后，立即确认不可路由的 mandatory 或 immediate 消息。当所有应该将消息路由到它们的队列都已收到传递确认或已将消息加入队列时，其他消息将被确认，必要时将消息持久化。

注：当 mandatory 标志位设置为 true 时，如果 exchange 根据自身类型和消息 routingKey 无法找到一个合适的 queue 存储消息，那么 broker 会调用 basic.return 方法将消息返还给生产者;当 mandatory 设置为 false 时，出现上述情况 broker 会直接将消息丢弃;通俗的讲，mandatory 标志告诉 broker 代理服务器至少将消息 route 到一个队列中，否则就将消息 return 给发送者;

当 noWait 为真时，客户端不会等待响应。如果服务端不支持此方法，则可能会发生通道异常。

具体代码实现如下：

...

// 设置消息确认
channel.Confirm(false)
confirms = channel.NotifyPublish(make(chan amqp.Confirmation, 1))

...

go func() {
		for confirm := range confirms {
			if confirm.Ack { // 消息已确认
				fmt.Printf("confirmed delivery with delivery tag: %d \n", confirm.DeliveryTag)
			} else { // 未确认的消息可以重新发送
				fmt.Printf("failed confirmed delivery of delivery tag: %d \n", confirm.DeliveryTag)
			}
		}
	}()

...

消费者可靠性

// 将autoAck设置为false
func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error) {
	// When we return from ch.call, there may be a delivery already for the
	// consumer that hasn't been added to the consumer hash yet.  Because of
	// this, we never rely on the server picking a consumer tag for us.

	if err := args.Validate(); err != nil {
		return nil, err
	}

	if consumer == "" {
		consumer = uniqueConsumerTag()
	}

	req := &basicConsume{
		Queue:       queue,
		ConsumerTag: consumer,
		NoLocal:     noLocal,
		NoAck:       autoAck,
		Exclusive:   exclusive,
		NoWait:      noWait,
		Arguments:   args,
	}
	res := &basicConsumeOk{}

	deliveries := make(chan Delivery)

	ch.consumers.add(consumer, deliveries)

	if err := ch.call(req, res); err != nil {
		ch.consumers.cancel(consumer)
		return nil, err
	}

	return (<-chan Delivery)(deliveries), nil
}

立即开始消费排队的消息。

在 Connection 或 Channel 上的任何其他操作之前开始接收返回的 chan Delivery。

消息会继续往返回的 chan Delivery 传递，直到发生 Channel.Cancel、Connection.Close、Channel.Close 或 AMQP 异常。消费者必须在 chan 范围内确保收到所有消息。未收到的消息将阻塞同一连接上的所有方法。

AMQP 中的所有消息都必须得到确认。消费者在成功处理消息后最好手动调用 Delivery.Ack。如果消费者被取消或通道或连接被关闭，任何未确认的消息将在同一队列的末尾重新入队。

消费者由一个字符串标识，该字符串是唯一的，适用于该 channal 上的所有消费者。如果希望最终取消消费者，请在 Channel.Cancel 中使用相同的非空标识符。空字符串将导致重新成唯一标识。消费者身份将包含在 ConsumerTag 字段中的每个消息中

当 autoAck（也称为 noAck）为真时，服务器将在将消息写入网络之前向该消费者确认确认。当 autoAck 为真时，消费者不应调用 Delivery.Ack。自动确认消息意味着如果服务器投递消息后消费者无法处理某些消息，则可能会丢失某些消息。

当 exclusive 为 true 时，服务器将确保这是该队列中的唯一消费者。当 exclusive 为 false 时，服务器将在多个消费者之间公平地分发消息。 RabbitMQ 不支持 noLocal 标志。建议对 Channel.Publish 和 Channel.Consume 使用单独的连接，以免在发布时 TCP 回推影响消费消息的能力，因此这里主要是为了完整性。当 noWait 为 true 时，不要等待服务器确认请求并立即开始消费。如果无法消费，则会引发通道异常并关闭通道。

消费消息时，将 autoAck 设置为 false

func (d Delivery) Ack(multiple bool) error {
	if d.Acknowledger == nil {
		return errDeliveryNotInitialized
	}
	return d.Acknowledger.Ack(d.DeliveryTag, multiple)
}

客户端消费到消息后，需要调用 ack 确认接收到消息

AMQP 中的所有消息的投递都必须得到确认。如果使用 autoAck true 调用 Channel.Consume，那么服务端将自动确认每条消息，但是不应该调用此方法，因为这个不能保证消费端业务处理成功。所以，必须在成功处理消息后调用 Delivery.Ack。当 multiple 为真时，此消息和同一通道上所有先前未确认的消息将被确认，这对于消息的批处理很有用（但是有个弊端就是，如果有一个出错了，所有批处理的数据都需要重发）。对于每个未自动确认的消息，都必须调用 Delivery.Ack、Delivery.Reject 或 Delivery.Nack。

消费端的确认机制的实现：

...

//读取数据
	if msgs, err = channel.Consume(queue.Name, "", false, false, false, false, nil); err != nil {
		fmt.Println("Consume Err =", err)
		return
	}
	go func() {
		for msg := range msgs {
			fmt.Println("Receive Msg =", string(msg.Body))
      // 确认消息
			msg.Ack(false)
		}
	}()
...