Go语言系列之RabbitMQ消息队列

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

1. RabbitMQ是什么？

MQ 是什么？队列是什么，MQ 我们可以理解为消息队列，队列我们可以理解为管道。以管道的方式做消息传递。

生活场景：

1.其实我们在双11的时候，当我们凌晨大量的秒杀和抢购商品，然后去结算的时候，就会发现，界面会提醒我们，让我们稍等，以及一些友好的图片文字提醒。而不是像前几年的时代，动不动就页面卡死，报错等来呈现给用户。

在这业务场景中，我们就可以采用队列的机制来处理，因为同时结算就只能达到这么多。

2.在我们平时的超市中购物也是一样，当我们在结算的时候，并不会一窝蜂一样涌入收银台，而是排队结算。这也是队列机制。

2. RabbitMQ简介

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。 RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。下面将重点介绍RabbitMQ中的一些基础概念，了解了这些概念，是使用好RabbitMQ的基础。

可靠性(Reliablity)：使用了一些机制来保证可靠性，比如持久化、传输确认、发布确认。
灵活的路由(Flexible Routing)：在消息进入队列之前，通过Exchange来路由消息。对于典型的路由功能，Rabbit已经提供了一些内置的Exchange来实现。针对更复杂的路由功能，可以将多个Exchange绑定在一起，也通过插件机制实现自己的Exchange。
消息集群(Clustering)：多个RabbitMQ服务器可以组成一个集群，形成一个逻辑Broker。
高可用(Highly Avaliable Queues)：队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
多种协议(Multi-protocol)：支持多种消息队列协议，如STOMP、MQTT等。
多种语言客户端(Many Clients)：几乎支持所有常用语言，比如Java、.NET、Ruby等。
管理界面(Management UI)：提供了易用的用户界面，使得用户可以监控和管理消息Broker的许多方面。
跟踪机制(Tracing)：如果消息异常，RabbitMQ提供了消息的跟踪机制，使用者可以找出发生了什么。
插件机制(Plugin System)：提供了许多插件，来从多方面进行扩展，也可以编辑自己的插件。

2.1 定义和特征

RbbitMQ是面向消息的中间件，用于组件之间的解耦，主要体现在消息的发送者和消费者之间无强依赖关系
RabbitMQ特点：高可用，可扩展，多语言客户端，管理界面等；
主要使用场景：流量削峰，异步处理，应用解耦等；

2.2 安装

ubuntu 的参照： https://gitee.com/zhangyafeii/rabbitmq
windows的参照：https://www.cnblogs.com/JustinLau/p/11738511.html
以下为centos7的安装过程

安装erlang

# centos7
wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.17/rabbitmq-server-3.7.17-1.el7.noarch.rpm
yum install epel-release
yum install unixODBC unixODBC-devel wxBase wxGTK SDL wxGTK-gl
rpm -ivh esl-erlang_22.0.7-1~centos~7_amd64.rpm

　安装rabbitmq

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.0/rabbitmq-server-3.8.0-1.el7.noarch.rpm
yum -y install socat
rpm -ivh rabbitmq-server-3.8.0-1.el7.noarch.rpm

　启动

chkconfig rabbitmq-server on                     # 开机启动
systemctl start rabbitmq-server.service          # 启动
systemctl stop rabbitmq-server.service    		# 停止
systemctl restart rabbitmq-server.service		# 重启
rabbitmqctl status							  # 查看状态
rabbitmq-plugins enable rabbitmq_management      # 启动Web管理器

修改配置

vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.0/ebin/rabbit.app

将：{loopback_users, [<<”guest”>>]}， 改为：{loopback_users, []}， 原因：rabbitmq从3.3.0开始禁止使用guest/guest权限通过   除localhost外的访问

systemctl restart rabbitmq-server.service    # 重启服务

3. RabbitMQ核心概念

Broker：标识消息队列服务器实体.
Virtual Host：虚拟主机。标识一批交换机、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个vhost本质上就是一个mini版的RabbitMQ服务器，拥有自己的队列、交换器、绑定和权限机制。vhost是AMQP概念的基础，必须在链接时指定，RabbitMQ默认的vhost是 /。
Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。
Queue：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
Banding：绑定，用于消息队列和交换机之间的关联。一个绑定就是基于路由键将交换机和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。
Channel：信道，多路复用连接中的一条独立的双向数据流通道。新到是建立在真实的TCP连接内地虚拟链接，AMQP命令都是通过新到发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说，建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念，以复用一条TCP连接。
Connection：网络连接，比如一个TCP连接。
Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
Consumer：消息的消费者，表示一个从一个消息队列中取得消息的客户端应用程序。
Message：消息，消息是不具名的，它是由消息头和消息体组成。消息体是不透明的，而消息头则是由一系列的可选属性组成，这些属性包括routing-key(路由键)、priority(优先级)、delivery-mode(消息可能需要持久性存储[消息的路由模式])等。

AMQP消息路由

AMQP中消息的路由过程和JMS存在一些差别。AMQP中增加了Exchange和Binging的角色。生产者把消息发布到Exchange上，消息最终到达队列并被消费者接收，而Binding决定交换器的消息应该发送到哪个队列。

Exchange类型

Exchange分发消息时，根据类型的不同分发策略有区别。目前共四种类型：direct、fanout、topic、headers(headers匹配AMQP消息的header而不是路由键(Routing-key)，此外headers交换器和direct交换器完全一致，但是性能差了很多，目前几乎用不到了。所以直接看另外三种类型。)。

direct

消息中的路由键(routing key)如果和Binding中的binding key一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配。

fanout

每个发到fanout类型交换器的消息都会分到所有绑定的队列上去。fanout交换器不处理该路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout类型转发消息是最快的。

topic

topic交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键(routing-key)和绑定键(bingding-key)的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符："#"和"*"。#匹配0个或多个单词，匹配不多不少一个单词。

4. RabbitMQ的运行模式

简单模式

创建实例

package RabbitMQ

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

//url格式  amqp://账号:密码@rabbitmq服务器地址:端口号/vhost
const MQURL = "amqp://zhangyafei:[email protected]:5672/imooc"

type RabbitMQ struct {
    conn    *amqp.Connection
    channel *amqp.Channel
    // 队列名称
    QueueName string
    //交换机
    Exchange string
    //key
    Key string
    // 连接信息
    Mqurl string
}

//创建结构体实例
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
    rabbitmq := &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, Mqurl: MQURL}
    var err error
    // 创建rabbitmq连接
    rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
    rabbitmq.failOnErr(err, "创建连接错误！")
    rabbitmq.channel, err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err, "获取channel失败！")
    return rabbitmq
}

//断开channel和connection
func (r *RabbitMQ) Destory() {
    r.channel.Close()
    r.conn.Close()
}

//错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
    if err != nil {
        log.Fatalf("%s:%s", message, err)
        panic(fmt.Sprintf("%s:%s", message, err))
    }
}

// 简单模式step1： 1.创建简单模式下的rabbitmq实例
func NewRabbitMQSimple(queueName string) *RabbitMQ {
    return NewRabbitMQ(queueName, "", "")
}

// 简单模式step2： 2.简单模式下生产
func (r *RabbitMQ) PublishSimple(message string) {
    // 1. 申请队列，如果队列不存在则自动创建，如果存在则跳过创建
    // 保证队列存在，消息能发送到队列中
    _, err := r.channel.QueueDeclare(
        r.QueueName,
        // 是否持久化
        false,
        // 是否为自动删除
        false,
        // 是否具有排他性
        false,
        // 是否阻塞
        false,
        // 额外属性
        nil,
    )
    if err != nil {
        fmt.Println(err)
    }
    // 2. 发送消息到队列中
    r.channel.Publish(
        r.Exchange,
        r.QueueName,
        // 如果为true,根据exchange类型和routekey规则，如果无法找到符合条件的队列，则会把发送的消息返回给发送者
        false,
        // 如果为true,当exchange发送消息到队列后发现队列上没有绑定消费者，则会把消息发还给发送者
        false,
        amqp.Publishing{ContentType: "text/plain", Body: []byte(message)},
    )
}

// 简单模式step3： 3.简单模式下消费
func (r *RabbitMQ) ConsumeSimple() {
    // 1. 申请队列，如果队列不存在则自动创建，如果存在则跳过创建
    // 保证队列存在，消息能发送到队列中
    _, err := r.channel.QueueDeclare(
        r.QueueName,
        // 是否持久化
        false,
        // 是否为自动删除
        false,
        // 是否具有排他性
        false,
        // 是否阻塞
        false,
        // 额外属性
        nil,
    )
    if err != nil {
        fmt.Println(err)
    }
    // 2. 接收消息
    msgs, err := r.channel.Consume(
        r.QueueName,
        // 用来区分多个消费者
        "",
        // 是否自动应答
        true,
        // 是否具有排他性
        false,
        // 如果为true，表示不能将同一个conn中的消息发送给这个conn中的消费者
        false,
        // 队列是否阻塞
        false,
        nil,
    )
    if err != nil {
        fmt.Println(err)
    }
    forever := make(chan bool)
    // 3. 启用协程处理消息
    go func() {
        for d := range msgs {
            // 实现我们要处理的逻辑函数
            log.Printf("Received a message: %s", d.Body)
        }
    }()
    log.Printf("[*] waiting for messages, to exit process CTRL+C")
    <-forever
}

创建实例

生产者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
    "fmt"
)

func main()  {
    rabbitmq := RabbitMQ.NewRabbitMQSimple("imoocSimple")
    rabbitmq.PublishSimple("hello imooc!")
    fmt.Println("发送成功")
}

mainSimplePublish.go

　　消费者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
)

func main()  {
    rabbitmq := RabbitMQ.NewRabbitMQSimple("imoocSimple")
    rabbitmq.ConsumeSimple()
}

mainSimpleReceive.go

工作模式

生产者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
    "fmt"
    "strconv"
    "time"
)

func main()  {
    rabbitmq := RabbitMQ.NewRabbitMQSimple("imoocSimple")
    for i := 0; i<= 100; i++ {
        rabbitmq.PublishSimple("hello imooc!" + strconv.Itoa(i))
        time.Sleep(1 * time.Second)
        fmt.Println(i)
    }
}

mainWorkPublish.go

　　消费者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
)

func main()  {
    rabbitmq := RabbitMQ.NewRabbitMQSimple("imoocSimple")
    rabbitmq.ConsumeSimple()
}

mainWorkReceive.go

订阅模式

创建实例

// 订阅模式下创建RabbitMQ实例
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
    rabbitmq := NewRabbitMQ("", exchangeName, "")
    var err error
    // 获取connection
    rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
    rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
    // 获取channel
    rabbitmq.channel, err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err, "failed to open a channel")
    return rabbitmq
}

// 订阅模式下生产
func (r *RabbitMQ) PublishPub(message string) {
    // 1. 尝试创建交换机
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        "fanout", // 广播类型
        true,     // 持久化
        false,    // 是否删除
        false,    // true表示这个exchange不可以被client用来推送消息的，仅用来进行exchange和exchange之间的绑定
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare a exchange")
    // 2. 发送消息
    err = r.channel.Publish(
        r.Exchange,
        "",
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(message),
        })
}

// 订阅模式消费端的代码
func (r *RabbitMQ) ReceiveSub() {
    // 1. 尝试创建交换机
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        "fanout", // 广播类型
        true,     // 持久化
        false,    // 是否删除
        false,    // true表示这个exchange不可以被client用来推送消息的，仅用来进行exchange和exchange之间的绑定
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare a exchange")
    // 2. 试探性创建队列
    q, err := r.channel.QueueDeclare(
        "", // 随机生产队列名称
        false,
        false,
        true,
        false,
        nil,
    )
    r.failOnErr(err, "failed to declare a queue")
    // 绑定队列到 exchange中
    err = r.channel.QueueBind(
        q.Name,
        "", // 在订阅模式下，这里的key为空
        r.Exchange,
        false,
        nil)
    // 消费消息
    messages, err := r.channel.Consume(
        q.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        fmt.Println(err)
    }
    forever := make(chan bool)
    // 3. 启用协程处理消息
    go func() {
        for d := range messages {
            // 实现我们要处理的逻辑函数
            log.Printf("Received a message: %s", d.Body)
        }
    }()
    log.Printf("[*] waiting for messages, to exit process CTRL+C")
    <-forever
}

创建实例

生产者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
    "fmt"
    "strconv"
    "time"
)

func main() {
    rabbitmq := RabbitMQ.NewRabbitMQPubSub("NewProduct")
    for i := 0; i <= 100; i++ {
        rabbitmq.PublishPub("订阅模式生产第" + strconv.Itoa(i) + "条数据")
        fmt.Println("订阅模式生产第" + strconv.Itoa(i) + "条数据")
        time.Sleep(1 * time.Second)
    }
}

mainPub.go

消费者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
)

func main() {
    rabbitmq := RabbitMQ.NewRabbitMQPubSub("NewProduct")
    rabbitmq.ReceiveSub()
}

mainSub.go

路由模式

创建实例、

// 路由模式下创建RabbitMQ实例
func NewRabbitMQRouting(exchangeName string, routingkey string) *RabbitMQ {
    // 创建RabbitMQ实例
    rabbitmq := NewRabbitMQ("", exchangeName, routingkey)
    var err error
    // 获取connection
    rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
    rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
    // 获取channel
    rabbitmq.channel, err = rabbitmq.conn.Channel()
    rabbitmq.failOnErr(err, "failed to open a channel")
    return rabbitmq
}

// 路由模式发送消息
func (r *RabbitMQ) PublishRouting(message string) {
    // 1. 尝试创建交换机
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        "direct", // 定向类型
        true,     // 持久化
        false,    // 是否删除
        false,    // true表示这个exchange不可以被client用来推送消息的，仅用来进行exchange和exchange之间的绑定
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare a exchange")
    // 2. 发送消息
    err = r.channel.Publish(
        r.Exchange,
        r.Key,
        false,
        false,
        amqp.Publishing{
            ContentType: "text/plain",
            Body:        []byte(message),
        })
}

// 路由模式消费端的代码
func (r *RabbitMQ) ReceiveRouting() {
    // 1. 试探性的创建交换机
    err := r.channel.ExchangeDeclare(
        r.Exchange,
        "direct", // 广播类型
        true,     // 持久化
        false,    // 是否删除
        false,    // true表示这个exchange不可以被client用来推送消息的，仅用来进行exchange和exchange之间的绑定
        false,
        nil,
    )
    r.failOnErr(err, "Failed to declare a exchange")
    // 2. 试探性创建队列
    q, err := r.channel.QueueDeclare(
        "", // 随机生产队列名称
        false,
        false,
        true,
        false,
        nil,
    )
    r.failOnErr(err, "failed to declare a queue")
    // 绑定队列到 exchange中
    err = r.channel.QueueBind(
        q.Name,
        r.Key, // 在订阅模式下，这里的key为空
        r.Exchange,
        false,
        nil)
    // 消费消息
    messages, err := r.channel.Consume(
        q.Name,
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        fmt.Println(err)
    }
    forever := make(chan bool)
    // 3. 启用协程处理消息
    go func() {
        for d := range messages {
            // 实现我们要处理的逻辑函数
            log.Printf("Received a message: %s", d.Body)
        }
    }()
    log.Printf("[*] waiting for messages, to exit process CTRL+C")
    <-forever
}

创建实例

生产者

package main

import (
    "RabbitMQ/RabbitMQ/RabbitMQ"
    "fmt"
    "strconv"
    "time"
)

func main()  {
    rabbit_imooc_one := RabbitMQ.NewRabbitMQRouting("exImooc", "imooc_one")
    rabbit_imooc_two := RabbitMQ.NewRabbitMQRouting("exImooc", "imooc_two")
    for i := 0; i <= 100; i++ {
        rabbit_imooc_one.PublishRouting("Hello imooc one!" + strconv.Itoa(i))
        rabbit_imooc_two.PublishRouting("Hello imooc two!" + strconv.Itoa(i))
        time.Sleep(1 * time.Second)
        fmt.Println(i)
    }
}

PublishRouting.go

　　消费者1

package main

import "RabbitMQ/RabbitMQ/RabbitMQ"

func main()  {
    rabbitmq_imooc_one := RabbitMQ.NewRabbitMQRouting("exImooc", "imooc_one")
    rabbitmq_imooc_one.ReceiveRouting()
}

ReceiveRouting1.go

　　消费者2

package main

import "RabbitMQ/RabbitMQ/RabbitMQ"

func main()  {
    rabbitmq_imooc_two := RabbitMQ.NewRabbitMQRouting("exImooc", "imooc_two")
    rabbitmq_imooc_two.ReceiveRouting()
}

ReceiveRouting2.go