1.安装依赖包

go get github.com/streadway/amqp

导入依赖包

import (
  "github.com/streadway/amqp"
)

2.发送消息

下面分步骤演示消息生产者如何完成消息推送

2.1. 连接RabbitMQ Server

// 连接RabbitMQ Server
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()

连接地址说明：amqp://账号:密码@RabbitMQ地址:端口/

2.2. 创建Channel

大部分操作都是在Channel（信道 ）完成的。

ch, err := conn.Channel()
defer ch.Close()

2.3. 声明队列

代表我们需要读写哪个队列

q, err := ch.QueueDeclare(
  "hello", // 队列名字
  false,   // 消息是否持久化
  false,   // 不使用的时候删除队列
  false,   // exclusive
  false,   // no-wait
  nil,     // arguments
)

2.4. 推送消息

// 消息内容
body := "Hello World!"

// 推送消息
err = ch.Publish(
  "",     // exchange（交换机名字），这里忽略
  q.Name, // 路由参数，这里使用队列名字作为路由参数
  false,  // mandatory
  false,  // immediate
  amqp.Publishing {
    ContentType: "text/plain",
    Body:        []byte(body),  // 消息内容
  })

2.5.发送消息完整代码

package main

// 导入包
import (
    "log"
    "github.com/streadway/amqp"
)

// 错误处理
func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    // 连接RabbitMQ
    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", // name
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

   // 要发送的消息内容
    body := "Hello World!"

    // 发送消息
    err = ch.Publish(
        "",     // 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", body)
}

3.接收消息

接收消息的前面三个步骤跟发送消息一样，分别对应2.1、2.2、2.3章节。 完整的接收消息代码如下:

package main

// 导入包
import (
    "log"
    "github.com/streadway/amqp"
)

// 错误处理
func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func main() {
    // 连接RabbitMQ
    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", // 队列名需要跟发送消息的队列名保持一致
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "Failed to declare a queue")

    // 创建消息消费者
    msgs, err := ch.Consume(
        q.Name, // 队列名
        "",     // 消费者名字，不填，则自动生成一个唯一ID
        true,   // 是否自动提交消息，即自动告诉rabbitmq消息已经处理成功。
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failOnError(err, "Failed to register a consumer")

    // 循环拉取队列中的消息
    for d := range msgs {
        // 打印消息内容
        log.Printf("Received a message: %s", d.Body)
    }
}

贡献者

sreio

更新日志

2025/7/7 09:20

查看所有更新日志

• b1a29-fixed于 2025/7/7
• ec6b6-queue于 2025/7/7