RabbitMQ实战:扩展RabbitMQ与系列总结

本系列是「RabbitMQ实战:高效部署分布式消息队列」书籍的总结笔记。

本篇是「RabbitMQ实战」系列的最后一篇,主要介绍RabbitMQ插件,了解如何安装和启用它,列举一些常用的插件,以及如何自定义。

在介绍之前,先总结下本系列的主要内容,把它们串起来。

系列总结

开篇时,这样定义过RabbitMQ:它是一个开源的消息代理和队列服务器,可以通过基本协议在完全不同的应用之间共享数据,可以将作业排队以便让分布式服务进行处理。

这句话有几个关键词:消息代理、队列服务器、共享数据、分布式处理,分别来说明下,以加深理解。

消息代理:可以把RabbitMQ看成一个代理服务器,一方面把消息生产者和消费者进行了解耦,更灵活;一方面,消息如何分发不用生产者考虑了,RabbitMQ提供多种分发策略。

队列服务器:消息最终是缓存在队列中的,消费者从指定的队列中消费消息,所以需要管理队列和队列中的消息。

共享数据:主要是说消息可以在不同应用间传输数据,通过AMQP协议进行规范,达到共享数据的目的。

分布式处理:多个消费者可以订阅同一个队列,这样,多台机器就可以同时处理同一批数据,达到分布式处理的效果。

如果理解AMQP基本元素和消息模型,上面的概念会很好理解,有2篇文章重点介绍了这块。

有些场景,对RabbitMQ可用性要求比较高,不容许有消息的丢失,需要了解RabbitMQ的可用性保障和实现。另外, 要能够实时监控RabbitMQ的本身及各个组件的运行情况,有问题时及时报警,快速进行处理。

如果消息量和并发量比较高,需要根据不同的业务场景,在可用性和性能上进行平衡,以满足业务的需要。

有些消息比较敏感,在传输的过程中要进行加密处理,通过ssl协议可以很好的解决。

最后,RabbitMQ提供了一些列的插件,为我们提供了很多有用的功能,比如Web管理界面插件、Shovel跨机房复制插件等,还允许我们自定义插件,扩展需要的功能。

完整索引

  1. 理解消息通信
  2. 运行和管理
  3. 消息通信模型和最佳实践
  4. 可用性分析和实现
  5. 界面管理和监控
  6. 数据传输的安全性介绍
  7. 性能和安全
  8. 扩展

插件介绍

当需要某些功能而服务器没有时,可以通过添加插件的方式来进行增强,可以从网络上找,也可以自己编写插件。

用插件可以做什么

先简单举几个例子,这些可以通过安装插件来解决:

  • 支持AMQP以外的协议;
  • 不同的认证机制(LDAP、自定义数据库);
  • 消息复制;
  • 新的交换器和路由算法;
  • 消息日志和审计;

STOMP是一个简单的基于文本的协议,用于在应用之间传输数据,它可以与ActiveMQ服务器一起工作,如果你的代码基于ActiveMQ和STOMP,但想使用RabbitMQ,就可以使用STOMP插件进行适配。

假设你的系统中所有的用户管理均通过LDAP,想在RabbitMQ中使用它进行认证,可以使用rabbitmq-auth-backend-ldap插件进行集成。

LDAP是轻量目录访问协议,一个为查询、浏览和搜索而优化的专业分布式数据库,它呈树状结构组织数据,就好象Linux/Unix系统中的文件目录一样。目录数据库和关系数据库不同,它有优异的读性能,但写性能差,并且没有事务处理、回滚等复杂功能,不适于存储修改频繁的数据。所以目录天生是用来查询的,就好象它的名字一样。

还可以自定义路由规则的交换器,下面会详细介绍。

查找和管理插件

可以查看 官网页面,查找到维护和实验阶段的插件。

再来看看插件的一些管理命令:

开启插件,插件开启后,需要重启RabbitMQ服务器:

1
./rabbitmq-plugins enable <plugin-name>

禁用插件,禁用插件后,需要重启RabbitMQ服务器:

1
./rabbitmq-plugins disable <plugin-name>

查看启用的插件:

1
./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交换器进行实现。

  • 需要缓存已路由的消息,每当交换器路由消息的时候,就将该消息存储到某个数据库中,该功能可在函数route/2中实现;
  • 当队列绑定到交换器时,需要把缓存的消息投递过去,该功能可在函数add_binding/3中实现;
  • 当交换器删除的时候,需要丢弃缓存的消息,该功能可在函数delete/3中实现;

然后,实现交换器behavious,我没有看相关实现细节,感兴趣的可以查看实现代码:传送门

最后,将交换器注册到RabbitMQ,RabbitMQ维护了一个注册表,用这张表来跟踪所有的交换器类型及其模块名称,假设将消息发布到fanout交换器上,RabbitMQ会进入注册表,检查由哪个模块来实现fanout交换器,找到后,会继续调用该模块的路由函数。

所以,需要找到一个方法将自定义交换器添加到注册表中,RabbitMQ支持启动步骤的概念,当服务器启动时会调用一系列步骤,可以在模块中添加一个启动步骤,RabbitMQ启动是,会将自定义交换器添加到rabbit_registry注册表中。

1
2
3
4
5
6
7
-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}]}).

以上,自定义交换器就完成了,可以编写程序进行验证。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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);

欢迎扫描下方二维码,关注我的个人微信公众号,分享我的工作、学习和生活 ~


情情说