本文共 3602 字,大约阅读时间需要 12 分钟。
本系列是「RabbitMQ实战:高效部署分布式消息队列」书籍的总结笔记。
本篇是「RabbitMQ实战」系列的最后一篇,主要介绍RabbitMQ插件,了解如何安装和启用它,列举一些常用的插件,以及如何自定义。
在介绍之前,先总结下本系列的主要内容,把它们串起来。
开篇时,这样定义过RabbitMQ:它是一个开源的消息代理和队列服务器,可以通过基本协议在完全不同的应用之间共享数据,可以将作业排队以便让分布式服务进行处理。
这句话有几个关键词:消息代理、队列服务器、共享数据、分布式处理,分别来说明下,以加深理解。
消息代理:可以把RabbitMQ看成一个代理服务器,一方面把消息生产者和消费者进行了解耦,更灵活;一方面,消息如何分发不用生产者考虑了,RabbitMQ提供多种分发策略。
队列服务器:消息最终是缓存在队列中的,消费者从指定的队列中消费消息,所以需要管理队列和队列中的消息。
共享数据:主要是说消息可以在不同应用间传输数据,通过AMQP协议进行规范,达到共享数据的目的。
分布式处理:多个消费者可以订阅同一个队列,这样,多台机器就可以同时处理同一批数据,达到分布式处理的效果。
如果理解AMQP基本元素和消息模型,上面的概念会很好理解,有2篇文章重点介绍了这块。
有些场景,对RabbitMQ可用性要求比较高,不容许有消息的丢失,需要了解RabbitMQ的可用性保障和实现。另外, 要能够实时监控RabbitMQ的本身及各个组件的运行情况,有问题时及时报警,快速进行处理。
如果消息量和并发量比较高,需要根据不同的业务场景,在可用性和性能上进行平衡,以满足业务的需要。
有些消息比较敏感,在传输的过程中要进行加密处理,通过ssl协议可以很好的解决。
最后,RabbitMQ提供了一些列的插件,为我们提供了很多有用的功能,比如Web管理界面插件、Shovel跨机房复制插件等,还允许我们自定义插件,扩展需要的功能。
完整索引:
当需要某些功能而服务器没有时,可以通过添加插件的方式来进行增强,可以从网络上找,也可以自己编写插件。
先简单举几个例子,这些可以通过安装插件来解决:
STOMP是一个简单的基于文本的协议,用于在应用之间传输数据,它可以与ActiveMQ服务器一起工作,如果你的代码基于ActiveMQ和STOMP,但想使用RabbitMQ,就可以使用STOMP插件进行适配。
假设你的系统中所有的用户管理均通过LDAP,想在RabbitMQ中使用它进行认证,可以使用rabbitmq-auth-backend-ldap插件进行集成。
LDAP是轻量目录访问协议,一个为查询、浏览和搜索而优化的专业分布式数据库,它呈树状结构组织数据,就好象Linux/Unix系统中的文件目录一样。目录数据库和关系数据库不同,它有优异的读性能,但写性能差,并且没有事务处理、回滚等复杂功能,不适于存储修改频繁的数据。所以目录天生是用来查询的,就好象它的名字一样。
还可以自定义路由规则的交换器,下面会详细介绍。
可以查看 ,查找到维护和实验阶段的插件。
再来看看插件的一些管理命令:
开启插件,插件开启后,需要重启RabbitMQ服务器:
./rabbitmq-plugins enable
禁用插件,禁用插件后,需要重启RabbitMQ服务器:
./rabbitmq-plugins disable
查看启用的插件:
./rabbitmq-plugins list -e
使用一个具体的例子来介绍自定义插件的开发,有这样一个场景:使用RabbitMQ为聊天应用建模,该模型中有一个全局聊天室,所有的用户都连接到这里,每位用户拥有自己的队列,绑定到全局fanout交换器上。
每次发消息到该交换器上时,该消息会群发给所有绑定的队列上,但如果有一个新的客户端连接到这个聊天室,只会得到发送给聊天室的新消息,无法了解在加入之前会话的上下文。
如果能将最近的消息发送给新来的客户,用户体验更更好点,可以通过自定义交换器实现这个功能。
RabbitMQ开发者制定了开发环境,即RabbitMQ Public Umbrella,通过把它从代码库中签出,然后把自己的插件添加到项目结构中。
RabbitMQ是用Erlang语言开发的,Erlang源代码是以模块的方式组织起来的,模块内的函数实现了应用程序需要提供的功能,插件只需要一个模块来包含自定义交互器的实现。
在面向对象编程中,拥有接口的概念,在Erlang中有相同的概念:behavious(行为)。
Erlang behavior 确定了模块需要实现和导出哪些函数,这样调用该模块的代码才知道该如何使用它。
另一个不同的地方在于,Erlang没有java那样可见性概念,拥有的就是一个模块导出列表,如果模块实现了函数fun1、fun2和fun3,但只导出fun1的话,fun2和fun3将无法被外界调用。
另外,还有一个函数参数数量的概念,可以接受名字相同、参数数量不同的函数,因此behavious可以确定函数fun1/1和fun1/2是两个不同的函数。
首先,确认要扩展的behavios,RabbitMQ暴露了一个交换器behavious,名为rabbit_exchange_type,它会明确需要实现哪些函数来成为一个符合要求的交换器。
然后,梳理实现思路:我们要实现的是一个加强版的fanout类型交换器,缓存最近20条消息,可以基于RabbitMQ上的fanout交换器进行实现。
然后,实现交换器behavious,我没有看相关实现细节,感兴趣的可以查看实现代码:。
最后,将交换器注册到RabbitMQ,RabbitMQ维护了一个注册表,用这张表来跟踪所有的交换器类型及其模块名称,假设将消息发布到fanout交换器上,RabbitMQ会进入注册表,检查由哪个模块来实现fanout交换器,找到后,会继续调用该模块的路由函数。
所以,需要找到一个方法将自定义交换器添加到注册表中,RabbitMQ支持启动步骤的概念,当服务器启动时会调用一系列步骤,可以在模块中添加一个启动步骤,RabbitMQ启动是,会将自定义交换器添加到rabbit_registry注册表中。
-rabbit_boot_step({rabbit_exchange_type_rh_registry,[{description, "recent history exchange type: registry"}, {mfa, {rabbit_registry, register, [exchange, <<"x-recent-history">>, ?MODULE]}}, //注册自定义交换器 {requires, rabbit_registry}, {enables, kernel_ready}]}).
以上,自定义交换器就完成了,可以编写程序进行验证。
require_once('../lib/php-amqplib/amqp.inc');define('HOST', 'localhost');define('PORT', 5672);define('USER', 'guest');define('PASS', 'guest');define('VHOST', '/');$exchange = 'rh-exchange';$conn = new AMQPConnection(HOST, PORT, USER, PASS, VHOST);$ch = $conn->channel();$ch->exchange_declare($exchange, 'x-recent-history', //使用自定义交换器 false, true, false);
欢迎扫描下方二维码,关注我的个人微信公众号,分享我的工作、学习和生活~
转载于:https://blog.51cto.com/13714880/2120199