0
0

RabbitMQ的六种消息模式（golang实现）

RabbitMQ 的主要作用

  1. 消息传递：- RabbitMQ 充当消息中介，将消息从生产者传递到消费者。它可以确保消息的可靠传输，即使接收者临时不可用，消息也能被妥善保存，等待消费。
  2. 异步处理：- 通过 RabbitMQ，应用程序可以将任务放入队列中，随后由一个或多个消费者异步处理。这种机制非常适合耗时较长的操作，如数据处理、文件转换等。
  3. 系统解耦：- RabbitMQ 使得生产者和消费者之间可以解耦。生产者只负责发送消息，而不关心消息的处理逻辑，消费者则可以独立地处理这些消息。
  4. 负载均衡：- RabbitMQ 支持多消费者的负载均衡模式，即同一个队列中的消息可以被多个消费者处理，从而实现任务的并行处理和系统的负载均衡。
  5. 消息路由：- RabbitMQ 提供了多种路由机制，通过不同类型的交换机（如 direct、fanout、topic、headers）将消息路由到合适的队列中，满足复杂的消息分发需求。

RabbitMQ 的优势

  1. 高可用性和可靠性：- RabbitMQ 支持持久化、镜像队列等机制，确保消息在系统故障或崩溃时不丢失，增强了消息传递的可靠性。
  2. 丰富的功能和灵活性：- RabbitMQ 提供了多种消息传递模式和高级特性，如消息确认、延迟队列、死信队列等，可以满足各种应用场景的需求。
  3. 可扩展性：- RabbitMQ 可以通过集群、分片和高可用性配置扩展其处理能力，支持从单节点到分布式集群的灵活部署，以满足不同规模的应用需求。
  4. 多语言支持：- RabbitMQ 支持多种编程语言，如 Go、Java、Python、Ruby、C# 等，使得它能够轻松集成到各种技术栈中。
  5. 社区支持和插件系统：- 作为一个成熟的开源项目，RabbitMQ 拥有活跃的社区支持，并提供丰富的插件来扩展其功能，如集成监控、管理 UI、流量控制等。
  6. 跨平台支持：- RabbitMQ 可以部署在各种操作系统上，如 Linux、Windows、macOS，同时支持在云环境中运行，如 AWS、Azure、Google Cloud 等。

典型使用场景

  • 异步任务处理：如电子邮件发送、图像处理、视频转换等后台任务。
  • 微服务间通信：在微服务架构中，RabbitMQ 可用于不同服务之间的事件通知和数据传输。
  • 事件驱动架构：通过 RabbitMQ，可以实现复杂的事件驱动系统，支持事件的广播和处理。
  • 日志收集：集中收集和处理分布式系统中的日志信息。
  • 实时消息传递：如即时消息应用、在线交易系统中的订单处理。

六种消息模式

Simple Work Queue（简单工作队列）

  • 这是最基本的模式，生产者将消息直接发送到队列，消费者从队列中取出消息进行处理
  • 适用于单一消费者的简单场景

Work Queue（工作队列）

  • 在此模式下，多个消费者从同一个队列中获取消息，实现任务的发布和负载均衡
  • 适用于需要多个消费者处理同一类型任务的场景

Publish/Subscribe（发布/订阅模式）

  • 生产者将消息发送到交换机，交换机将消息广播到与之绑定的所有队列，所有订阅的消费者都能收到消息
  • 适用于广播类型的消息，如通知或更新

Routing（路由模式）

  • 消息生产者将消息发送到交换机，并为消息指定一个路由键。交换机根据路由键将消息发送到与路由键匹配的队列
  • 适用于消息需要根据条件选择性地发送到不同场景

Topic（主题模式）

  • 与路由模式类似，但路由键可以是模糊匹配（如使用通配符）。交换机将消息发送到符合匹配规则的队列
  • 适用于消息需要更灵活的路由规则的场景，如日志分发系统

Remote Procedure Call（RPC模式）

  • 使用RabbitMQ来实现远程过程调用（RPC）。生产者发送请求消息，消费者处理请求并将结果发送回生产者
  • 适用于需要同步调用和返回结果场景

四种交换机（Exchange）

direct （直连交换机）

  • 将队列绑定到交换机，消息的 routeKey 需要与队列绑定的 routeKey 相同

Fanout（扇形交换机）

  • 不处理 routeKey ，直接把消息转发到与其绑定的所有队列中

Topic（主题交换机）

  • 根据一定的规则，根据 routeKey 把消息转发到符合规则的队列中，其中 # 用于匹配符合一个或者多个词（范围更广）， * 用于匹配一个词

Headers（头部交换机）

  • 根据消息的 headers 转发消息而不是根据 routeKey 来转发消息, 其中 header 是一个 Map，也就意味着不仅可以匹配字符串类型，也可以匹配其他类型数据。 规则可以分为所有键值对匹配或者单一键值对匹配
    消息模式交换机简单工作队列、工作队列空交换机发布/订阅模式fanout（扇形交换机）路由模式direct （直连交换机）主题模式topic（主题交换机）RPC模式direct （直连交换机）

    Simple Work Queue（简单工作队列）

生产者将消息直接发送到队列，消费者从队列中取出消息进行处理

生产者代码

package main
// 生产者代码 send.goimport("context"
    amqp "github.com/rabbitmq/amqp091-go""log""time")// failOnError 处理错误使用，后续不再描述funcfailOnError(err error, msg string){if err !=nil{// 打印 错误日志
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 1、建立mq连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 2、建立通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明一个队列
    q, err := ch.QueueDeclare("hello",// name 队列名称false,// durable 是否持久化false,// delete when unused  队列不再使用就删除false,// exclusive 队列是否仅供当前连接使用false,// no-wait 声明队列时是否等待RabbitMQ确认nil,// arguments 传递额外的队列参数)failOnError(err,"Failed to declare a queue")// 创建一个带有超时的上下文，设置超时时间为5秒。在发送消息时，如果操作超时，上下文会取消操作
    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()// 消息体
    body :="Hello World!"// 用于将消息发布到指定的交换机
    err = ch.PublishWithContext(ctx,"",// exchange 交换机
        q.Name,// routing key 路由键false,// mandatory 消息是否必须路由到队列false,// immediate
        amqp.Publishing{
            ContentType:"text/plain",
            Body:[]byte(body),})failOnError(err,"Failed to publish a message")
    log.Printf(" [x] Sent %s\n", body)}

消费者代码

package main
// 消费者代码 receive.goimport("log"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 创建通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明队列
    q, err := ch.QueueDeclare("hello",// namefalse,// durablefalse,// delete when unusedfalse,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")// 注册消费者
    msgs, err := ch.Consume(
        q.Name,// queue 队列名称"",// consumer 消费者的标签，通常用来标识不同的消费者实例true,// auto-ack 表示是否自动确认消息。这里设置为 true，即消费者在接收到消息后会自动确认false,// exclusive 队列是否仅供当前连接使用。这里设置为 false，即队列可以被多个消费者共享false,// no-local 是否排除本地发送的消息。这里设置为 false，即本地发送的消息也会被消费者接收false,// no-waitnil,// args)failOnError(err,"Failed to register a consumer")// 定义一个空的通道 var forever chanstruct{}// 启动一个新的协程，用于处理从队列中接收到的消息gofunc(){for d :=range msgs {
            log.Printf("Received a message: %s", d.Body)}}()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")<-forever
}

执行

# 启动生产者
go run send.go
2024/08/29 14:16:55  [x] Sent Hello World, 你好!# 启动消费者
go run receive.go
2024/08/29 14:16:50  [*] Waiting for messages. To exit press CTRL+C
2024/08/29 14:16:55 Received a message: Hello World, 你好!

Work Queue（工作队列）

通过将任务分发到多个工作者进行处理，能实现负载均衡、异步处理和容错性。它非常适合于后台任务处理、异步工作、分布式计算等场景。使用RabbitMQ或类似的消息队列系统，可以轻松地实现工作队列模式，提升系统的处理能力和效率。

img

生产者代码

package main
// 生产者代码 send.goimport("context""log""os""strings""time"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明队列
    q, err := ch.QueueDeclare("task_queue",// nametrue,// durablefalse,// delete when unusedfalse,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")// 创建上下文
    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()// 获取消息体
    body :=bodyFrom(os.Args)
    err = ch.PublishWithContext(ctx,"",// exchange 交换机
        q.Name,// routing key route keyfalse,// mandatoryfalse,
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,// 设置消息的持久性。amqp.Persistent 表示消息是持久的
            ContentType:"text/plain",
            Body:[]byte(body),})failOnError(err,"Failed to publish a message")
    log.Printf(" [x] Sent %s", body)}funcbodyFrom(args []string)string{var s stringif(len(args)<2)|| os.Args[1]==""{
        s ="hello"}else{
        s = strings.Join(args[1:]," ")}return s
}

消费者代码

package main
//消费者代码 work.goimport("bytes""log""time"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()

    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()

    q, err := ch.QueueDeclare("task_queue",// nametrue,// durablefalse,// delete when unusedfalse,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")

    err = ch.Qos(1,// prefetch count 表示在消费者发送 ack 确认之前，RabbitMQ 允许同一消费者最多接收多少条未确认的消息0,// prefetch size 以字节为单位，指定消费者在发送 ack 之前可以接收的消息总大小。这里设置为 0，表示没有限制false,// global 如果设置为 true，则 prefetch count 和 prefetch size 将对整个通道（所有消费者）生效。如果设置为 false，则仅对当前消费者生效)failOnError(err,"Failed to set QoS")// 注册消费者
    msgs, err := ch.Consume(
        q.Name,// queue"",// consumerfalse,// auto-ackfalse,// exclusivefalse,// no-localfalse,// no-waitnil,// args)failOnError(err,"Failed to register a consumer")var forever chanstruct{}gofunc(){for d :=range msgs {
            log.Printf("Received a message: %s", d.Body)
            dotCount := bytes.Count(d.Body,[]byte("."))// 统计点数
            t := time.Duration(dotCount)// 计算睡眠时间
            time.Sleep(t * time.Second)
            log.Printf("Done")
            d.Ack(false)// 确认消息，向RabbitMQ发送一个确认信号，表示该消息已被成功处理}}()

    log.Printf(" [*] Waiting for messages. To exit press CTRL+C")<-forever
}

执行

# shell1 启动消费者1
go run work.go
2024/08/29 14:55:53  [*] Waiting for messages. To exit press CTRL+C
2024/08/29 14:55:58 Received a message: 12024/08/29 14:55:58 Done
2024/08/29 14:56:03 Received a message: 32024/08/29 14:56:03 Done
2024/08/29 14:56:10 Received a message: 5# shell2 启动消费者2
go run work.go 
2024/08/29 14:55:56  [*] Waiting for messages. To exit press CTRL+C
2024/08/29 14:56:00 Received a message: 22024/08/29 14:56:00 Done
2024/08/29 14:56:06 Received a message: 42024/08/29 14:56:06 Done

# shell3 启动生产者，传入不同参数
go run send.go 1
go run send.go 2
go run send.go 3
go run send.go 4
go run send.go 5

Publish/Subscribe（发布/订阅模式）

发布/订阅模式是一种非常灵活的消息传递模式，适用于需要将消息广播到多个订阅者的场景。它解耦了消息的生产者和消费者，支持多种交换机类型和路由策略，使得系统具有高扩展性和灵活性。在需要广播、事件通知或实时数据推送的应用场景中，发布/订阅模式是一种理想的选择。

发布者代码

package main
// 发布者代码 publish.goimport("context""log""os""strings""time"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs",// name 交换机名称"fanout",// type 交换机类型true,// durable 持久化表示false,// auto-deleted 消息自动删除false,// internalfalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare an exchange")

    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()

    body :=bodyFrom(os.Args)// 用于将消息发布到指定的交换机
    err = ch.PublishWithContext(ctx,"logs",// exchange 交换机名称"",// routing keyfalse,// mandatoryfalse,// immediate
        amqp.Publishing{
            ContentType:"text/plain",
            Body:[]byte(body),})failOnError(err,"Failed to publish a message")

    log.Printf(" [x] Sent %s", body)}// 获取启动命令行参数funcbodyFrom(args []string)string{var s stringif(len(args)<2)|| os.Args[1]==""{
        s ="hello"}else{
        s = strings.Join(args[1:]," ")}return s
}

订阅者代码

package main
// 订阅者代码 subscribe.goimport("log"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs",// name"fanout",// typetrue,// durablefalse,// auto-deletedfalse,// internalfalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare an exchange")// 建立队列
    q, err := ch.QueueDeclare("",// namefalse,// durablefalse,// delete when unusedtrue,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")// 队列绑定（订阅）交换机
    err = ch.QueueBind(
        q.Name,// queue name"",// routing key"logs",// exchangefalse,nil,)failOnError(err,"Failed to bind a queue")// 接收 消息
    msgs, err := ch.Consume(
        q.Name,// queue"",// consumertrue,// auto-ackfalse,// exclusivefalse,// no-localfalse,// no-waitnil,// args)failOnError(err,"Failed to register a consumer")var forever chanstruct{}gofunc(){for d :=range msgs {
            log.Printf(" [x] %s", d.Body)}}()

    log.Printf(" [*] Waiting for logs. To exit press CTRL+C")<-forever
}

执行，先执行订阅者，在执行发布者

# shell1 启动订阅者一
go run subscribe.go
2024/08/30 10:41:11  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 10:42:11  [x] 震惊！拼多多大跳水
2024/08/30 10:42:41  [x] 一分钟速成
# shell2 启动订阅者二
go run subscribe.go
2024/08/30 10:41:11  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 10:42:11  [x] 震惊！拼多多大跳水
2024/08/30 10:42:41  [x] 一分钟速成

#shell3 启动发布者
go run publish.go 震惊！拼多多大跳水
2024/08/30 10:42:11  [x] Sent 震惊！拼多多大跳水
go run publish.go 一分钟速成
2024/08/30 10:42:41  [x] Sent 一分钟速成

Routing（路由模式）

路由模式通过路由键和Direct Exchange实现了对消息的精确控制，允许消息定向发送到指定的队列。这种模式非常适合需要对消息进行分类处理的应用场景，如日志系统、任务分发、报警系统等。通过路由模式，可以确保每个消费者只接收和处理与其相关的消息，从而提高系统的灵活性和效率。

img

生产者代码

// 生产者代码 send.gopackage main

import("context""log""os""strings""time"

    amqp "github.com/rabbitmq/amqp091-go")funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 闯进通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs_direct",// name 交换机名称"direct",// type 交换机类型 直连交换机true,// durablefalse,// auto-deletedfalse,// internalfalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare an exchange")

    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()

    body :=bodyFrom(os.Args)
    err = ch.PublishWithContext(ctx,"logs_direct",// exchangeseverityFrom(os.Args),// routing keyfalse,// mandatoryfalse,// immediate
        amqp.Publishing{
            ContentType:"text/plain",
            Body:[]byte(body),})failOnError(err,"Failed to publish a message")

    log.Printf(" [x] Sent %s", body)}funcbodyFrom(args []string)string{var s stringif(len(args)<3)|| os.Args[2]==""{
        s ="hello"}else{
        s = strings.Join(args[2:]," ")}return s
}funcseverityFrom(args []string)string{var s stringif(len(args)<2)|| os.Args[1]==""{
        s ="info"}else{
        s = os.Args[1]}return s
}funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}

消费者代码

package main

// 消费者代码 work.goimport("log""os"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 创建通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs_direct",// name 交换机名称"direct",// type 交换机类型true,// durable 持久化false,// auto-deletedfalse,// internalfalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare an exchange")// 声明队列
    q, err := ch.QueueDeclare("",// name 队列名称false,// durable 持久化false,// delete when unusedtrue,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")iflen(os.Args)<2{
        log.Printf("Usage: %s [info] [warning] [error]", os.Args[0])
        os.Exit(0)}for_, s :=range os.Args[1:]{
        log.Printf("Binding queue %s to exchange %s with routing key %s",
            q.Name,"logs_direct", s)// 绑定交换机
        err = ch.QueueBind(
            q.Name,// queue name 队列名
            s,// routing key"logs_direct",// exchangefalse,nil)failOnError(err,"Failed to bind a queue")}// 接受消息
    msgs, err := ch.Consume(
        q.Name,// queue"",// consumertrue,// auto ackfalse,// exclusivefalse,// no localfalse,// no waitnil,// args)failOnError(err,"Failed to register a consumer")var forever chanstruct{}gofunc(){for d :=range msgs {
            log.Printf(" [x] %s", d.Body)}}()

    log.Printf(" [*] Waiting for logs. To exit press CTRL+C")<-forever
}

执行代码

## 先启动消费者## shell1 接受routing key 为info的消息
go run work.go info
2024/08/30 11:54:32 Binding queue amq.gen-QYGiiUUDVkpiiOwCOTcfJQ to exchange logs_direct with routing key info
2024/08/30 11:54:32  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 11:56:09  [x] 你好，欢迎~

## shell2 接受routing key 为warn、error的消息
go run work.go warn error
2024/08/30 11:54:55 Binding queue amq.gen-dhxRepsx9kZY_TssKNq60Q to exchange logs_direct with routing key warn
2024/08/30 11:54:55 Binding queue amq.gen-dhxRepsx9kZY_TssKNq60Q to exchange logs_direct with routing key error
2024/08/30 11:54:55  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 11:56:27  [x] 发出警告
2024/08/30 11:56:42  [x] 系统错误

## shell3 启动生产者
go run send.go info 你好，欢迎~
2024/08/30 11:56:09  [x] Sent 你好，欢迎~

go run send.go warn 发出警告
2024/08/30 11:56:27  [x] Sent 发出警告

go run send.go error 系统错误
2024/08/30 11:56:42  [x] Sent 系统错误

Topic（主题模式）

主题模式通过使用通配符和主题交换机，实现了灵活且强大的消息路由功能。它特别适合需要对消息进行复杂过滤和分类的场景，如日志系统、通知系统和事件驱动架构。在这些场景中，主题模式能够根据复杂的条件灵活地分发消息，确保消费者只接收和处理与其相关的消息。

主题模式使用通配符（‘***’ 和 '#**’）进行消息路由。生产者发布的消息携带一个路由键（Routing Key），而消费者的队列通过绑定键（Binding Key）订阅与其匹配的消息。通配符允许绑定键匹配一组路由键，从而实现灵活的消息分发。

‘****’：匹配单个单词。例如，绑定键为 log. 时，可以匹配 log.info、log.error 等路由键。

‘**#**’：匹配零个或多个单词。例如，绑定键为 log.# 时，可以匹配 log.info、log.error、log.warn.high 等所有以 log. 开头的路由键。

img

生产者代码

package main

// 生产者 send.goimport("context"
    amqp "github.com/rabbitmq/amqp091-go""log""os""strings""time")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs_topic",// name 交换机名称"topic",// type 交换机类型 topic 主题模式true,// durable 持久化false,// auto-deleted 当所有与交换机绑定的队列都被删除时，交换机会自动删除。如果设置为 false，交换机不会自动删除。false,// internal 如果设置为 true，交换机将是内部的，这意味着它不能被客户端直接使用，只能通过其他交换机进行路由。通常设置为 falsefalse,// no-wait ：如果设置为 true，表示不等待服务器返回结果。交换机声明后，立即继续执行后续代码，不等待确认。如果设置为 false，程序会等待服务器的确认，确保交换机成功声明nil,// arguments)failOnError(err,"Failed to declare an exchange")

    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()

    body :=bodyFrom(os.Args)
    err = ch.PublishWithContext(ctx,"logs_topic",// exchangeseverityFrom(os.Args),// routing keyfalse,// mandatory 如果设置为 true，则表示如果消息无法路由到任何队列，RabbitMQ 将返回该消息给发布者。如果设置为 false，消息将被丢弃false,// immediate 如果设置为 true，表示如果消息发送时，队列中没有消费者，消息将不会入队列，并返回给发布者。一般情况下，这个选项很少使用，通常设置为 false
        amqp.Publishing{
            ContentType:"text/plain",
            Body:[]byte(body),})failOnError(err,"Failed to publish a message")

    log.Printf(" [x] Sent %s", body)}funcbodyFrom(args []string)string{var s stringif(len(args)<3)|| os.Args[2]==""{
        s ="hello"}else{
        s = strings.Join(args[2:]," ")}return s
}funcseverityFrom(args []string)string{var s stringif(len(args)<2)|| os.Args[1]==""{
        s ="anonymous.info"}else{
        s = os.Args[1]}return s
}

消费者代码

package main

// 消费者 work.goimport("log""os"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明交换机
    err = ch.ExchangeDeclare("logs_topic",// name 交换机名称"topic",// type 交换机类型true,// durable 持久化false,// auto-deleted 当所有与交换机绑定的队列都被删除时，交换机会自动删除。如果设置为 false，交换机不会自动删除。false,// internal 如果设置为 true，交换机将是内部的，这意味着它不能被客户端直接使用，只能通过其他交换机进行路由。通常设置为 falsefalse,// no-wait ：如果设置为 true，表示不等待服务器返回结果。交换机声明后，立即继续执行后续代码，不等待确认。如果设置为 false，程序会等待服务器的确认，确保交换机成功声明nil,// arguments)failOnError(err,"Failed to declare an exchange")// 声明队列
    q, err := ch.QueueDeclare("",// name 队列名称 mq 自动生成false,// durable 持久化false,// delete when unusedtrue,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")iflen(os.Args)<2{
        log.Printf("Usage: %s [binding_key]...", os.Args[0])
        os.Exit(0)}for_, s :=range os.Args[1:]{
        log.Printf("Binding queue %s to exchange %s with routing key %s",
            q.Name,"logs_topic", s)// 交换机绑定队列
        err = ch.QueueBind(
            q.Name,// queue name 队列名
            s,// routing key"logs_topic",// exchangefalse,nil)failOnError(err,"Failed to bind a queue")}

    msgs, err := ch.Consume(
        q.Name,// queue"",// consumertrue,// auto ackfalse,// exclusivefalse,// no localfalse,// no waitnil,// args)failOnError(err,"Failed to register a consumer")var forever chanstruct{}gofunc(){for d :=range msgs {
            log.Printf(" [x] %s", d.Body)}}()

    log.Printf(" [*] Waiting for logs. To exit press CTRL+C")<-forever
}

执行

## 先启动消费者## shell1 '#'表示接收所有消息
go run work.go #2024/08/30 14:07:26 Binding queue amq.gen--Qe9R8VFRrL5ZPwP-9MwRg to exchange logs_topic with routing key #2024/08/30 14:07:26  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 14:07:30  [x] 欢迎来到本系统
2024/08/30 14:07:50  [x] 存在一个警告！
2024/08/30 14:08:28  [x] 发生了一个错误！！！
2024/08/30 14:08:59  [x] 错误解除了~

## shell2 接受warn开头或者error结尾的消息
go run work.go warn.* *.error
2024/08/30 14:07:22 Binding queue amq.gen-bZoWyLRq4GL9KD-Ecl_60w to exchange logs_topic with routing key warn.*
2024/08/30 14:07:22 Binding queue amq.gen-bZoWyLRq4GL9KD-Ecl_60w to exchange logs_topic with routing key *.error
2024/08/30 14:07:22  [*] Waiting for logs. To exit press CTRL+C
2024/08/30 14:07:50  [x] 存在一个警告！
2024/08/30 14:08:28  [x] 发生了一个错误！！！

## 启动生产者## shell3 发送消息
go run send.go info.qwe 欢迎来到本系统
2024/08/30 14:07:30  [x] Sent 欢迎来到本系统
go run send.go warn.system 存在一个警告！
2024/08/30 14:07:50  [x] Sent 存在一个警告！
go run send.go gateway.error 发生了一个错误！！！
2024/08/30 14:08:28  [x] Sent 发生了一个错误！！！
go run send.go info.info 错误解除了~
2024/08/30 14:08:59  [x] Sent 错误解除了~

Remote Procedure Call（RPC模式）

RPC 模式通过模拟本地方法调用，实现了远程服务之间的高效通信。在分布式系统、微服务架构和跨语言调用等场景中，RPC 模式极大地简化了服务之间的交互，同时确保了请求和响应的匹配和处理。RabbitMQ 提供了可靠的消息传递机制，使得 RPC 模式更加健壮和可扩展。

客户端-服务器模型， 在 RPC 模式中，客户端发出请求，服务器端执行相应的操作并返回结果。这种模式模拟了本地函数调用的方式，使得远程服务调用看起来像是本地调用

请求/响应机制， 客户端发送请求消息，服务器处理后返回响应消息。每个请求都期望一个对应的响应，这种一对一的消息模式确保了客户端能够收到并处理来自服务器的返回值。

唯一的Correlation ID，每个请求都带有一个唯一的 Correlation ID，服务器处理完成后将该 ID 返回给客户端。客户端通过 Correlation ID 来匹配请求和响应，从而确保多个并发请求的结果不会混淆。

客户端代码

package main

// 客户端代码 client.goimport("context""log""math/rand""os""strconv""strings""time"

    amqp "github.com/rabbitmq/amqp091-go")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}// randomString 获取随机字符串funcrandomString(l int)string{
    bytes :=make([]byte, l)for i :=0; i < l; i++{
        bytes[i]=byte(randInt(65,90))}returnstring(bytes)}// 获取随机数funcrandInt(min int, max int)int{return min + rand.Intn(max-min)}// 调用服务请求结果funcfibonacciRPC(n int)(res int, err error){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明队列
    q, err := ch.QueueDeclare("",// namefalse,// durablefalse,// delete when unusedtrue,// exclusivefalse,// noWaitnil,// arguments)failOnError(err,"Failed to declare a queue")// 接收消息
    msgs, err := ch.Consume(
        q.Name,// queue"",// consumertrue,// auto-ackfalse,// exclusivefalse,// no-localfalse,// no-waitnil,// args)failOnError(err,"Failed to register a consumer")// 唯一的Correlation ID
    corrId :=randomString(32)

    ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()// 发送消息
    err = ch.PublishWithContext(ctx,"",// exchange"rpc_queue",// routing keyfalse,// mandatoryfalse,// immediate
        amqp.Publishing{
            ContentType:"text/plain",
            CorrelationId: corrId,
            ReplyTo:       q.Name,
            Body:[]byte(strconv.Itoa(n)),})failOnError(err,"Failed to publish a message")for d :=range msgs {if corrId == d.CorrelationId {// 解析结果
            res, err = strconv.Atoi(string(d.Body))failOnError(err,"Failed to convert body to integer")break}}return}funcmain(){// 随机序列
    rand.Seed(time.Now().UTC().UnixNano())

    n :=bodyFrom(os.Args)

    log.Printf(" [x] Requesting fib(%d)", n)
    res, err :=fibonacciRPC(n)failOnError(err,"Failed to handle RPC request")

    log.Printf(" [.] Got %d", res)}funcbodyFrom(args []string)int{var s stringif(len(args)<2)|| os.Args[1]==""{
        s ="30"}else{
        s = strings.Join(args[1:]," ")}
    n, err := strconv.Atoi(s)failOnError(err,"Failed to convert arg to integer")return n
}

服务端代码

package main

// 服务端代码 server.goimport("context"
    amqp "github.com/rabbitmq/amqp091-go""log""strconv""time")funcfailOnError(err error, msg string){if err !=nil{
        log.Panicf("%s: %s", msg, err)}}// fib 求斐波拉契funcfib(n int)int{if n ==0{return0}elseif n ==1{return1}else{returnfib(n-1)+fib(n-2)}}funcmain(){// 建立连接
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")failOnError(err,"Failed to connect to RabbitMQ")defer conn.Close()// 打开通道
    ch, err := conn.Channel()failOnError(err,"Failed to open a channel")defer ch.Close()// 声明队列
    q, err := ch.QueueDeclare("rpc_queue",// namefalse,// durablefalse,// delete when unusedfalse,// exclusivefalse,// no-waitnil,// arguments)failOnError(err,"Failed to declare a queue")

    err = ch.Qos(1,// prefetch count 表示在消费者发送 ack 确认之前，RabbitMQ 允许同一消费者最多接收多少条未确认的消息0,// prefetch size 以字节为单位，指定消费者在发送 ack 之前可以接收的消息总大小。这里设置为 0，表示没有限制false,// global 如果设置为 true，则 prefetch count 和 prefetch size 将对整个通道（所有消费者）生效。如果设置为 false，则仅对当前消费者生效)failOnError(err,"Failed to set QoS")// 接收消息
    msgs, err := ch.Consume(
        q.Name,// queue"",// consumerfalse,// auto-ackfalse,// exclusivefalse,// no-localfalse,// no-waitnil,// args)failOnError(err,"Failed to register a consumer")var forever chanstruct{}gofunc(){
        ctx, cancel := context.WithTimeout(context.Background(),5*time.Second)defercancel()for d :=range msgs {// 获取消息
            n, err := strconv.Atoi(string(d.Body))failOnError(err,"Failed to convert body to integer")

            log.Printf(" [.] fib(%d)", n)// 计算结果
            response :=fib(n)// 发送消息
            err = ch.PublishWithContext(ctx,"",// exchange
                d.ReplyTo,// routing keyfalse,// mandatoryfalse,// immediate
                amqp.Publishing{
                    ContentType:"text/plain",
                    CorrelationId: d.CorrelationId,
                    Body:[]byte(strconv.Itoa(response)),})failOnError(err,"Failed to publish a message")// 确认消息已被成功使用
            d.Ack(false)}}()

    log.Printf(" [*] Awaiting RPC requests")<-forever
}

执行

## 先启动服务端，在启动客户端# shell1 
go run client.go
2024/08/30 14:41:53  [x] Requesting fib(30)2024/08/30 14:41:54  [.] Got 832040# shell2
go run server.go
2024/08/30 14:41:48  [*] Awaiting RPC requests
2024/08/30 14:41:54  [.] fib(30)

点击获取本篇文章源码

标签： rabbitmq golang 分布式
本文转载自: https://blog.csdn.net/weixin_46337064/article/details/141956201
版权归原作者 小镇cxy 所有， 如有侵权，请联系我们删除。
登录后发布评论

“RabbitMQ的六种消息模式（golang实现）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

[linux]Ubuntu 18.04安装arm-linux-gcc交叉编译器的两种方法

Kafka-生产者源码分析

Docker 安装 PostgreSQL

重学SpringBoot3-Spring Retry实践

上海乐鑫科技总代理商ESP32-C5，2.4&5GHz双频Wi-Fi6，高效连接更安全

15种常见的 Web 安全攻击类型以及防御手段

Ubuntu 22.04 不登录或锁屏时不能远程连接图形界面

文章导航