当前位置：首页 > news >正文

RabbitMQ-默认读、写方式介绍

news 2025/8/4 3:49:43

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的方法，后面再介绍其他几种使用方式。

查看全文

http://www.lryc.cn/news/353834.html