消息队列原理之rabbitmq-低调大师

消息队列原理之rabbitmq

2020-12-20 569

导读

谈起消息队列，我们的脑海可能会不由自主的冒出这么几个关键词，解耦、异步化、消峰、广播等，消息队列的种类也很多，如 rabbitmq、rocketmq、activemq、kafka等还有各个云厂商提供的消息队列。它们都有各种的特点和使用场景，所以这个系列的文章主要谈各个消息的原理，目前规划了两篇文章，rabbitmq 和 kafka ，其他的暂时还没有用到，还没有深究。

这篇主要介绍 rabbitmq 的原理和基于 golang 如何使用。

本文首发于 https://russellgao.cn/mq-rabbitmq/

介绍

RabbitMQ 是一个由 Erlang 开发的 AMQP(Advanced Message Queuing Protocol，高级消息队列协议)的开源实现，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。支持多种客户端语言。

架构

整体架构对照下面的图说明

先看看图片上各个名次的解释:

Broker:它提供一种传输服务，它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输，简单来说就是消息队列服务器实体。
Connection: 客户端与 Rabbitmq Broker 直接的 TCP 连接，通常一个客户端与 Broker 之间只需要一个连接即可。
Channel: 消息通道，在客户端的每个连接里，可建立多个channel，最好每个线程都用独立的Channel，后续的对 Queue 和 Exchange 的操作都是在 Channel 中完成的。
Producer: 消息生产者，通过和 Broker 建立 Connection 和 Channel ，向 Exchange 发送消息。
Consumer: 消息消费者，通过和 Broker 建立 Connection 和 Channel，从 Queue 中消费消息。
Exchange: 消息交换机，按照一定的策略把 Producer 生产的消息投递到 Queue 中，等待消费者消费。
Queue: 消息队列载体，每个消息都会被投入到一个或多个队列。
Vhost: 虚拟主机，一个broker里可以开设多个vhost，用作权限分离，把不同的系统使用的rabbitmq区分开，共用一个消息队列服务器，但看上去就像各自在用不用的rabbitmq服务器一样。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。
RoutingKey：路由关键字，生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。

这里面比较难理解的概念是 RoutingKey,Exchange,Binding ，消费发送时不会直接发送给 Queue ,而是先发送给 Exchange，由 Exchange 按照一定的规则投递到与它绑定的 Queue 中，那这个规则是什么呢? 规则就与 Exchange 的 Type、Binding、RoutingKey 相关，Exchange 支持的类型有 4 种，direct,fanout,topic,headers,含义如下:

direct: Queue 和 Exchange 在绑定时需要指定一个 key, 我们称为 Bindkey。Producer 往 Exchange 发送消息时，也需要指定一个 key ，这个 key 就是 Routekey。这种模式下 Exchange 会把消息投递给 Routekey 和 Bindkey 相同的队列

fanout: 类似于广播的方式，会把消息投递给和 Exchange 绑定的所有队列，不需要检查 Routekey 和 Bindkey 。

topic: 类似于组播的方式，这种模式下 Bingkey 支持模糊匹配，* 代表匹配一个任意词组，#代表匹配0个或多个词组。如 Producer 产生一条 RouteKey 为 benz.car 的消息，同时这个 Exchange 绑定了3组队列（请注意是3组不是3个，意思是Exchange可以和同一个Queue进行多次绑定，通过Bindkey 的不同，它们之间是多对多的关系），Bindkey 分别为: car ,*.car ,benz.car ,那么会把这个消息投递到 *.car、benz.car 对应的 Queue 中。

headers: 这个类型 Routekey 和 Bindkey 的匹配规则来路由消息，而是根据发送的消息内容中的 headers 属性进行匹配。

对照上面图和名次解释应该比较清晰明了了，下面我们通过几个例子说明如何使用。

用法（golang）

direct

先看看 Rabbitmq 默认的 exchange ，其中第一个(AMQP default) 是默认的，默认绑定了所有的 Queue ，会把消息投递到 Routekey 对应的队列中，即： Routekey==QueueName 。

package main

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

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

var url = "amqp://username:password@ip:port"

func main() {
	conn, err := amqp.Dial(url)
	handlerError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

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

	queueNameCar := "car"
	if _, err := channel.QueueDeclare(queueNameCar, false, false, false, false, nil); err != nil {
		handlerError(err, "Failed to decare Queue")
	}

	if err := channel.Publish("", queueNameCar, false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}
}

这里是一个完整的 Demo，后面只会提供main() 函数的示例代码，其他的和这里这里类似。
申明了一个名称为 car 的消息队列，并没有做任何的绑定，往 defalut exchange 发送一条消息，routekey 为 car ,可以看到和队列名相同。
为了方便演示，结果以图片的方式展现，可以看到这里有 car 的队列，并且有一条消息。

在创建队列有几个参数可以关注一下

Durability：持久化，是否将队列持久化到磁盘，当选择持久化时当 rabbitmq 重启了，这个队列还在，否则当重启了之后这个队列就没有了，需要重新创建，这个需要设计程序时考虑到。

Auto delete: 当其中一个消费者已经完成之后，会删除这个队列并断开与其他的消费者的连接。

Arguments：

x-message-ttl: 消息的过期时间，发布到队列中的消息在被丢弃之前可以存活多久。

x-expires: 队列的过期时间，一个队列在多长时间内未使用会被自动删除。

x-max-length: 队列的长度，最多剋容纳多少条消息。

x-max-length-bytes: 队列最大可以包含多大的消息。

x-dead-letter-exchange: 当消息过期或者被客户端reject 之后应该重新投递到那个exchange ，类似与一个producer发送消息时选择exchange

x-dead-letter-routing-key: 当消息过期或者被客户端reject 之后重新投递时的 Routekey，类似与一个producer发送消息时设置routekey，默认是原消息的 routekey

x-max-priority: 消息的优先级设置，设置可以支持的最大优先级，如设置为10,则可以在发送消息设置优先级，可以根据优先级处理消息，默认为空，当为空时则不支持优先级

x-queue-mode: 将队列设置为懒惰模式，尽可能多地将消息保留在磁盘上，以减少RAM的使用量；如果不设置，队列将保留内存中的缓存，以尽可能快地传递消息。

我们自己创建一个 direct 类型的 exchange 并绑定一些队列看看是什么效果。

func main() {
	conn, err := amqp.Dial(url)
	handlerError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

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

	directExchangeNameCar := "direct.car"
	if err := channel.ExchangeDeclare(directExchangeNameCar, "direct", true, false, false, false, nil); err != nil {
		handlerError(err, "Failed to decalare exchange")
	}

	queueNameCar := "car"
	queueNameBigCar := "big-car"
	queueNameMiddleCar := "middle-car"
	queueNameSmallCar := "small-car"
	channel.QueueDeclare(queueNameCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameBigCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameMiddleCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameSmallCar, false, false, false, false, nil)

	if err := channel.QueueBind(queueNameCar, "car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameBigCar, "car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameBigCar, "big.car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.QueueBind(queueNameMiddleCar, "car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameMiddleCar, "middler.car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.QueueBind(queueNameSmallCar, "car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameSmallCar, "small.car", directExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.Publish(directExchangeNameCar, "car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}
}

代码中申明了 1 一个 Exchange ，4个 Queue，7个 Binding ,其中一个 Binding 详情如下:
可以看到向这个 Exchange 中发消息，Routekey 为 car ,匹配的队列有个，那么这4个队列中都应该有消息才对和我们的设想是一直

Queue 的创建上面已经讲过了，这里有 Exchange 的创建，那么再看看创建 Exchange 有哪些参数

Type: 类型，上面已经涉及到了

Durability: 持久化

Auto delete: 是否自动删除，如果为yes 则当其中队列完成 unbind 操作，则其他的 queue 或者 exchange 也会 unbind 并且删除这个 exchange 。

Internal: 如果为yes ，则客户端不能直接往这个 exchange 上发送消息，只能用作和 exchange 绑定。

fanout

fanout 工作方式类似于广播，看看下面的代码

func main() {
	conn, err := amqp.Dial(url)
	handlerError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

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

	fanoutExchangeNameCar := "fanout.car"
	if err := channel.ExchangeDeclare(fanoutExchangeNameCar, "fanout", true, false, false, false, nil); err != nil {
		handlerError(err, "Failed to decalare exchange")
	}

	queueNameCar := "car"
	queueNameBigCar := "big-car"
	queueNameMiddleCar := "middle-car"
	queueNameSmallCar := "small-car"
	channel.QueueDeclare(queueNameCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameBigCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameMiddleCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameSmallCar, false, false, false, false, nil)

	if err := channel.QueueBind(queueNameCar, "car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameBigCar, "car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameBigCar, "big.car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.QueueBind(queueNameMiddleCar, "car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameMiddleCar, "middler.car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.QueueBind(queueNameSmallCar, "car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}
	if err := channel.QueueBind(queueNameSmallCar, "small.car", fanoutExchangeNameCar, false, nil); err != nil {
		handlerError(err, "Failed to bind queue to exchange")
	}

	if err := channel.Publish(fanoutExchangeNameCar, "middle.car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}
}

这个申明了一个 fanout 类型的 exchange ，和上面的代码类似，只有 exchange 不同。
可以先在脑海中想想每个 queue 中有几条消息。
向 fanout.car 这个 exchange 发消息指定 Routekey 为 middle.car ，但是由于是广播模式，所以和 routekey 是没有关系的，每个消息队列中各有一条消息。
请注意有些 binding 指向的是同一个 queue ，那么会产生多条消息到相同的 queue 中，答案是否定的。producer 产生一条消息，根据一定的规则，每个队列只会收到一条(如何符合投递规则的话)。

topic

topic 比较有意思了，和之前的简单粗暴的用法不一样了，先看看下面的代码，声明了一个 topic 类型的 exchange， 4个 queue

func main() {
	conn, err := amqp.Dial(url)
	handlerError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

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

	topicExchangeNameCar := "topic.car"
	if err := channel.ExchangeDeclare(topicExchangeNameCar, "topic", true, false, false, false, nil); err != nil {
		handlerError(err, "Failed to decalare exchange")
	}

	queueNameCar := "car"
	queueNameBigCar := "big-car"
	queueNameMiddleCar := "middle-car"
	queueNameSmallCar := "small-car"
	channel.QueueDeclare(queueNameCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameBigCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameMiddleCar, false, false, false, false, nil)
	channel.QueueDeclare(queueNameSmallCar, false, false, false, false, nil)

	if err := channel.QueueBind(queueNameCar, "car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameBigCar, "car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameBigCar, "big.car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }

    if err := channel.QueueBind(queueNameMiddleCar, "car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameMiddleCar, "middler.car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }

    if err := channel.QueueBind(queueNameSmallCar, "car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameSmallCar, "small.car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameSmallCar, "*.small.car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
    if err := channel.QueueBind(queueNameSmallCar, "#.small.car", topicExchangeNameCar, false, nil); err != nil {
        handlerError(err, "Failed to bind queue to exchange")
    }
}

现在思考每个 producer 产生消息之后，会有哪些 queue 会收到消息。

	if err := channel.Publish(topicExchangeNameCar, "car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}

每个 queue 都会收到消息

	if err := channel.Publish(topicExchangeNameCar, "small.car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}

small-car 这一个队列会收到消息。
- 符合 Routekey 为 small.car 、*.small.car、#.small.car 的binding

	if err := channel.Publish(topicExchangeNameCar, "benz.small.car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}

small-car 这一个队列会收到消息。
- 符合 Routekey 为 *.small.car、#.small.car 的binding

	if err := channel.Publish(topicExchangeNameCar, "auto.blue.benz.small.car", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}

small-car 这一个队列会收到消息。
- 符合 Routekey 为 #.small.car 的binding

	if err := channel.Publish(topicExchangeNameCar, "bike", false, false, amqp.Publishing{ContentType: "text/plain", Body: []byte("test car")}); err != nil {
		handlerError(err, "Failed to publish message")
	}

都不会收到消息，没有符合的 routekey 。

headers

这种类型很少有实际的应用场景。

参考

https://www.rabbitmq.com/documentation.html

微信关注我们

原文链接：https://my.oschina.net/russellgao/blog/4817171

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

从零开发HarmonyOS（鸿蒙）手机小游戏——数字华容道

HarmonyOS（鸿蒙）手机第一个小游戏app——数字华容道前言概述正文创建项目实现初始界面布局实现数字的随机打乱实现滑动或点击调换数字实现游戏成功界面源代码结语前言 2月16号HarmonyOS2.0手机开发者Beta版已经发布了，作为“1+8+N”战略的重要入口和生态核心，怎么能少得了手机应用开发呢，今天将由深鸿会深大学习小组（Zzt_01-23）从零基础开发第一个HarmonyOS手机小游戏——数字华容道（界面略丑陋，大佬别喷），此前已经在运动手表上成功开发了：HarmonyOS运动手表游戏合并、HarmonyOS手表游戏——数字华容道，同样是深鸿会深大小组学习完HarmonyOS后自行开发的一个鸿蒙demo，详细讲述了数字华容道在鸿蒙手机上开发思路。深鸿会深大学习小组是一群热衷于学习鸿蒙相关知识和开发鸿蒙相关应用的开发者们，我们的学习项目为：荔园Harmony，同时也欢迎与各位感兴趣的读者一起学习HarmonyOS开发，相互交流、共同进步。概述本个demo将从零基础开始完成鸿蒙小游戏APP在手机上的编译在项目中我们所使用到的软件为DevEco Stu...

2020-12-20

485

前言众所周知，NodeJS作为后端开发语言和运行环境，样样都好，就差一个NodeJS工作流引擎。CabloyJS 4.0重点开发了NodeJS工作流引擎，并作为内置的基础核心模块，近一步拓展了NodeJS在后端的应用场景，为深入研发各类商业业务逻辑，提供了基础支撑 NodeJS工作流引擎的特点更简便的配置：采用JSON进行流程定义的配置，告别XML配置文件的冗杂流程定义：支持历史版本、支持启用/禁用更清晰的架构：采用三个核心模块用分层的机制实现工作流引擎的架构，让工作流不再神秘，源码也不再叠床架屋模块名称说明 a-flow 流程定义、流程实例 a-flownode 流程节点（活动节点） a-flowtask 流程任务支持业务流程和审批流程与Atom三生三世结合，内置了一套基于Atom的审批工作流。参见：原子阶段（三生三世）与表单验证结合，支持分别配置不同流程节点的读取字段权限和修改字段权限。参见：表单验证可通过AOP机制定制工作流逻辑可通过Listener机制定制工作流逻辑开放式的架构，支持更多流程节点的定制开发包含大量测试驱动代码，可快速上手使用工作流工作...

2020-12-20

521

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。