simple 模式是RabbitMQ 里最基础的一种模式，其他模式是类似于 simple 模式，参数配置变化，发送和接受者稍作改动。

模型图

simple 模式的模型图很简单。
生产(Producing)的意思就是发送。发送消息的程序就是一个生产者(producer)。

队列(queue)就是存在于RabbitMQ中邮箱的名称。虽然消息的传输经过了RabbitMQ和你的应用程序，但是它只能被存储于队列当中。实质上队列就是个巨大的消息缓冲区，它的大小只受主机内存和硬盘限制。多个生产者（producers）可以把消息发送给同一个队列，同样，多个消费者（consumers）也能够从同一个队列（queue）中获取数据。

消费（Consuming）和接收(receiving)是同一个意思。一个消费者（consumer）就是一个等待获取消息的程序。

image.png
这里直接给出代码, 代码里有对应的注释
新建 rabbitmq.go

  1. package rabbitmq
  3. import (
  4.     "fmt"
  5.     "github.com/streadway/amqp"
  6.     "log"
  7. )
  9. // 连接信息
  10. const MQURL = "amqp://guest:guest@localhost:5672/"
  12. // rabbitMQ 连接体
  13. type RabbitMQ struct {
  14.     conn *amqp.Connection
  15.     channel *amqp.Channel
  16.     QueueName string    // 队列名称
  17.     Exchange string        // 交换机名称
  18.     Key string            // bind key 名称
  19.     MqUrl string        // 连接信息
  20. }
  22. // 创建RabbitMQ结构体实例
  23. func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
  24.     return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, MqUrl: MQURL}
  25. }
  27. // 断开 channel 和 connection
  28. func (r *RabbitMQ) Destroy() {
  29.     r.channel.Close()
  30.     r.conn.Close()
  31. }
  33. // 错误处理函数
  34. func (r *RabbitMQ) failOnErr(err error, message string) {
  35.     if err != nil {
  36.         log.Fatalf("[err]%s, [message]%s", err, message)
  37.     }
  38. }
  40. // 创建简单模式下RabbitMQ实例
  41. func NewRabbitMQSimple(queueName string) *RabbitMQ {
  42.     rmq := NewRabbitMQ(queueName, "", "")
  43.     var err error
  44.     // 获取 connection
  45.     rmq.conn, err = amqp.Dial(rmq.MqUrl)
  46.     rmq.failOnErr(err, "创建连接错误")
  47.     rmq.channel, err = rmq.conn.Channel()
  48.     rmq.failOnErr(err, "创建连接错误")
  49.     return rmq
  50. }
  52. func (r *RabbitMQ) PublishSimple(message string) {
  53.     // 1.申请队列, 如果队列不存在会自动创建, 如果存在则跳过创建
  54.     // 保证队列存在, 消息能发送到队列中
  55.     _, err := r.channel.QueueDeclare(
  56.         // name
  57.         r.QueueName,
  58.         // durable, 是否持久化
  59.         false,
  60.         // autoDelete, 是否为自动化删除
  61.         false,
  62.         // exclusive, 是否具有排他性
  63.         false,
  64.         // noWait, 是否阻塞
  65.         false,
  66.         // args,
  67.         nil,
  68.         )
  70.     if err != nil {
  71.         fmt.Println(err)
  72.     }
  74.     // 2.发送消息到队列中
  75.     r.channel.Publish(
  76.         r.Exchange,
  77.         r.QueueName,
  78.         // 如果为true, 根据 exchange 类型和 routekey 规则,
  79.         // 如果无法找到该队列，那么会把发送的消息返回给发送者。
  80.         false,
  81.         // 如果为ture时, 当 exchange 发送消息到队列后发现队列上没有绑定消费者,
  82.         // 这会把消息返回给消费者。
  83.         false,
  84.         amqp.Publishing{
  85.             ContentType: "text/plain",
  86.             Body: []byte(message),
  87.         },
  88.         )
  89. }
  91. func (r *RabbitMQ) ConsumeSimple() {
  92.     // 1.申请队列, 如果队列不存在会自动创建, 如果存在则跳过创建
  93.     // 保证队列存在, 消息能发送到队列中
  94.     _, err := r.channel.QueueDeclare(
  95.         r.QueueName,
  96.         // 是否持久化
  97.         false,
  98.         // 是否为自动化删除
  99.         false,
  100.         // 是否具有排他性
  101.         false,
  102.         // 是否阻塞
  103.         false,
  104.         nil,
  105.     )
  107.     if err != nil {
  108.         fmt.Println(err)
  109.     }
  111.     msgs, err := r.channel.Consume(
  112.         r.QueueName,
  113.         // 用来区分多个消费者
  114.         "",
  115.         // 是否自动应答, 默认为true; 为false 需要手动实现反馈回调
  116.         true,
  117.         // 是否具有排他性,
  118.         false,
  119.         // 如果设置为true, 表示不能将同一个conenction中发送的消息传递给这个connection中的消费者
  120.         false,
  121.         // 队列消费是否阻塞, false 表示阻塞。
  122.         false,
  123.         nil,
  124.         )
  125.     if err != nil {
  126.         fmt.Println(err)
  127.     }
  128.     forever := make(chan bool)
  130.     // 启动协程处理消息
  131.     go func() {
  132.         for d := range msgs {
  133.             // 实现我们要处理的逻辑函数
  134.             log.Printf("Receive a message: %s", d.Body)
  135.         }
  136.     }()
  137.     log.Printf("[*] Waiting for message. To exit to press CTRL+C")
  138.     <-forever
  139. }

我来测试下
send.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "github.com/WenkeZhou/flash-sale/pkg/rabbitmq"
  6. )
  8. func main() {
  9.     rmq := rabbitmq.NewRabbitMQSimple("simple")
  10.     rmq.PublishSimple("hello this is at test!")
  11.     fmt.Println("发送成功")
  12. }

receive.go

  1. package main
  3. import (
  4.     "fmt"
  5.     "github.com/WenkeZhou/flash-sale/pkg/rabbitmq"
  6. )
  8. func main() {
  9.     rmq := rabbitmq.NewRabbitMQSimple("simple")
  10.     rmq.ConsumeSimple()
  11.     fmt.Println("发送成功")
  12. }