RabbitMQ 学习笔记3 - 使用amqp库连接Rabbit

2020-04-02 本文已影响0人张云飞Vir

0. 背景

使用Go 操作RabbitMQ 收发消息，可以使用Go RabbitMQ客户端库连接 RabbitMQ 来实现。

1. amqp 类库介绍

amqp 类库是使用Go 操作 RabbitMQ 的一个 Go RabbitMQ客户端

在安装好 RabbitMQ 服务端后，就可以使用 Go 开发客户端程序来连接RabbitMQ，发送消息，或者取消息。

开始之前
要连接使用 RabbitMQ 首先要了解 AMQP 协议的基本概念，我的另一篇文章
做了介绍，本文末也有一些 AMQP的一些资源。

一些基本概念：

生产者：一个发送消息的程序，它产生消息并发送到队列。这里是用Go写的发送端示程序例。
消息队列：即 RabbitMQ 内部的队列，它安装在一个服务器中。做为消息中间件，它与具体开发语言无关，支持 Go,Java等接入连接。
消费者：消费者是一个等待消息，接收消息的接收端程序示例

消息队列

本文编写两个示例：

发送端：一个生产者发送一条消息
接收端：以及一个接收消息并打印出来的消费者。

2. 开始使用

2.1 发送端

新建一个文件 send.go，编写 go 代码。

步骤分解如下：
（1）建立连接
conn, err := amqp.Dial("amqp://admin:admin@dev.com:5672/")

（2）打开channel
这里的channel 是AMQP 里的概念，可以理解为多路复用的一个tcp长连接。

（3）声明一个队列
q, err := ch.QueueDeclare( ... )

（4）创建消息
msg := amqp.Publishing{ ... }

（5）发布消息
err = ch.Publish( ... ）

代码如下：

package main

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

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

func main() {
    // 连接 RabbitMQ
    conn, err := amqp.Dial("amqp://admin:admin@dev.com:5672/")
    failOnError(err, "连接失败")
    defer conn.Close()

    // 建立一个 channel ( 其实就是TCP连接 ）
    ch, err := conn.Channel()
    failOnError(err, "打开通道失败")
    defer ch.Close()

    // 创建一个名字叫 "hello" 的队列
    q, err := ch.QueueDeclare(
        "hello", // name
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "创建队列失败")

    // 构建一个消息
    body := "Hello World!"
    msg := amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    }

    // 构建一个生产者，将消息 放入队列
    err = ch.Publish(
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,  // immediate
        msg)
    log.Printf(" [x] Sent %s", body)
    failOnError(err, "Failed to publish a message")
}

2.2 接收端

步骤分解如下：
（1）建立连接
conn, err := amqp.Dial("amqp://admin:admin@dev.com:5672/")

（2）打开channel
这里的channel 是AMQP 里的概念，可以理解为多路复用的一个tcp长连接。

（3）声明一个队列
q, err := ch.QueueDeclare( ... )

（4）构建一个消费者
msgChan, err := ch.Consume( ... )

（5）不断的读取消息

for d := range msgChan {
log.Printf("收到消息: %s", d.Body)
}

新建一个文件 receive.go，编写 go 代码。

package main

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

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

func main() {
    // 连接 RabbitMQ
    conn, err := amqp.Dial("amqp://admin:admin@dev.com:5672/")
    failOnError(err, "连接失败")
    defer conn.Close()

    // 建立一个 channel ( 其实就是TCP连接 ）
    ch, err := conn.Channel()
    failOnError(err, "打开通道失败")
    defer ch.Close()

    // 创建一个名字叫 "hello" 的队列
    q, err := ch.QueueDeclare(
        "hello", // name
        false,   // durable
        false,   // delete when unused
        false,   // exclusive
        false,   // no-wait
        nil,     // arguments
    )
    failOnError(err, "创建队列失败")

    // 开启一个 消费者
    // 返回值是 ch 类型
    msgChan, err := ch.Consume(
        q.Name, // queue
        "",     // consumer
        true,   // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failOnError(err, "注册消费者 ，失败")

    //帮助阻塞
    forever := make(chan bool)

    // 开启一个 go程
    go func() {
        for d := range msgChan {
            log.Printf("收到消息: %s", d.Body)
        }
    }()

    log.Printf(" 等待消息...")
    <-forever
}