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服务器：

./rabbitmq-plugins enable <plugin-name>

禁用插件，禁用插件后，需要重启RabbitMQ服务器：

./rabbitmq-plugins disable <plugin-name>

查看启用的插件：

./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注册表中。

-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);

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

​

 

评论

1 楼 情情说 2018-05-04  

喜欢的可以关注我的微信公众号：情情说