1、RabbitMQ简介
rabbitmq是一个开源的消息中间件，主要有以下用途，分别是：
- 应用解耦：通过使用RabbitMQ，不同的应用程序之间可以通过消息进行通信，从而降低应用程序之间的直接依赖性，提高系统的可维护性、扩展性和容错性。
- 异步提速：通过将耗时的操作转化为异步执行，可以提高系统的响应速度和吞吐量，提升用户体验。
- 削峰填谷：在高峰时段，RabbitMQ可以缓存大量的消息，从而避免系统崩溃，并在低峰时段处理这些消息，提高系统的稳定性。
- 消息分发：RabbitMQ可以将消息分发到多个消费者进行处理，从而提高系统的灵活性和处理能力。
了解rabbitmq的设计架构，对理解mq如何使用有很大的帮助。
一个非常重要的点，mq中的生产者从来不是直接将消息发送到队列中的，而是将消息发送到了mq的交换机中(上图中的exchange为交换机)， 甚至生产者都不知道这条消息将被发送到哪个队列中。
交换机是个怎样的设计呢，他的一侧连接生产者，从生产者接收消息，另外一侧连接队列，将消息push进队列中，将消息push进一个队列，还是多个队列，还是抛弃，这些策略是由交换机的类型决定的，对于交换机的使用，后面详细介绍。
2、RabbitMQ安装
rabiitmq的安装，最简单的一种方式为运行mq的docker镜像，一行命令搞定：
# latest RabbitMQ 3.13
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management执行命令后，可以看到如下打印，则代表RabbitMQ启动成功：
镜像启动成功后，可以通过ip:15672打开mq控制台：
http://xxx.xx.xxx.xx:15672/#/
mq安装完成后，下面就可以进行实践啦。
3、默认模式读、写mq
rabbitmq官方的库：github.com/rabbitmq/amqp091-go
生产者侧代码：
package mainimport ("context""fmt""time"amqp "github.com/rabbitmq/amqp091-go"
)func Send(msg string) error {// 连接rabbitmqconn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")if err != nil {fmt.Println("connect error:", err)return err}defer conn.Close()// 创建通道ch, err := conn.Channel()if err != nil {fmt.Println("channel error:", err)return err}defer ch.Close()// 创建队列,使用默认的交换机q, err := ch.QueueDeclare("lp_default", // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // noWaitnil, // arguments)if err != nil {fmt.Println("queue declare error:", err)return err}ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)defer cancel()fmt.Println(q.Name)// body := "Hello World!"err = ch.PublishWithContext(ctx,"", // exchange,默认交换机q.Name, // routing keyfalse, // mandatoryfalse, // immediateamqp.Publishing{ContentType: "text/plain",Body: []byte(msg),})if err != nil {fmt.Println("publish error:", err)return err}return nil
}func main() {Send("Hello world")
}
运行上面代码后，可以在rabbitmq的客户端 看到这个队列：
点击队列，进入队列详情：
第一个框中，显式了队列详情，可以看出，这个队列绑定的是默认的交换机。
第二个框，点击后，可以看到队列中的消息详情。
消费者：
package mainimport ("fmt"amqp "github.com/rabbitmq/amqp091-go"
)func main() {conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")if err != nil {fmt.Println("connect error:", err)return}defer conn.Close()ch, err := conn.Channel()if err != nil {fmt.Println("Channel error:", err)return}defer ch.Close()q, err := ch.QueueDeclare("lp_default", // nametrue, // durablefalse, // delete when unusedfalse, // exclusivefalse, // no-waitnil, // arguments)if err != nil {fmt.Println("Queue Declare error:", err)return}msgs, err := ch.Consume(q.Name, // queue"", // consumertrue, // auto-ackfalse, // exclusivefalse, // no-localfalse, // no-waitnil, // args)if err != nil {fmt.Println("Consume error:", err)return}var forever chan struct{}go func() {for d := range msgs {fmt.Printf("Received a message: %s\n", d.Body)}}()fmt.Printf(" [*] Waiting for messages. To exit press CTRL+C")<-forever
}
代码运行记录：
liupeng@192 default % go run recive.go[*] Waiting for messages. To exit press CTRL+CReceived a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world
Received a message: Hello world在以上消费端代码中，如果代码在处理消息的过程中出现异常导致了程序退出，这样正在处理的这条消息就会丢失，为了避免这种情况的发生，rabbitmq设计了消息应答的机制，我们修改上面程序，将auto-ack参数设置为false，当处理完消息后，使用d.Ack(false)发送消息应答。
msgs, err := ch.Consume(q.Name, // queue"", // consumerfalse, // auto-ack，设置为false，取消自动应答false, // exclusivefalse, // no-localfalse, // no-waitnil, // args)if err != nil {fmt.Println("Consume error:", err)return}var forever chan struct{}go func() {for d := range msgs {fmt.Printf("Received a message: %s\n", d.Body)d.Ack(false) // 手动应答}}()fmt.Printf(" [*] Waiting for messages. To exit press CTRL+C")<-forever如果忘记了进行消息应答，消息会被重新发入调度队列，这样就会吃掉越来越多的内存。
但是，当rabbitmq的服务down掉后，队列中的消息仍然会丢失，为了保证在这种情况下，消息仍然能够不丢失，我们需要做两件事：队列不丢失+消息不丢失，代码如下：
队列持久化：
q, err := ch.QueueDeclare("hello", // nametrue, // durable，设置队列持久化false, // delete when unusedfalse, // exclusivefalse, // no-waitnil, // arguments
)
failOnError(err, "Failed to declare a queue")消息持久化：
将DeliveryMode设置为amqp.Persistent
err = ch.PublishWithContext(ctx,"", // exchange,默认交换机q.Name, // routing keyfalse, // mandatoryfalse, // immediateamqp.Publishing{ContentType: "text/plain",Body: []byte(msg),DeliveryMode: amqp.Persistent,})以上就是默认读写rabbitmq的方法，后面再介绍其他几种使用方式。
