淘宝开源分布式消息中间件Metamorphosis

最近，淘宝开源了分布式消息中间件 Memorphosis项目，它是Linkedin开源MQ——Kafka的Java版本，针对淘宝内部应用做了定制和优化。

据了解，Metamorphosis（以下简称Meta）的设计原则包括：

• 消息都是持久的，保存在磁盘。
• 吞吐量第一。
• 消费状态保存在客户端。
• 分布式，生产者、服务器和消费者都可分布。

Metamorphosis的总体 架构图如下：

除了完整实现Kafka的功能之外，淘宝开发团队还为Meta加入了额外的功能，使得Meta成为一个更为强大的通用消息中间件，包括：

• 彻底用Java重写的实现，高效的协议和通讯框架。
• 发送端的负载均衡。
• Master/Slave异步和同步复制的高可用方案。
• 专门用于广播消息的客户端实现。
• 与diamond结合使用的顺序发送消息功能。
• 支持事务，包括本地事务和分布式事务，实现JTA规范。

消息中间件中有两个角色：消息生产者和消息消费者。Meta里同样有这两个概念，消息生产者负责创建消息并发送到Meta服务器，Meta服务器会将消息持久化到磁盘，消息消费者从Meta服务器拉取消息并提交给应用消费。

从Meta的 技术手册中，我们可以更加深入的看到Meta的实现细节和应用指南。

在使用消息生产者和消费者之前，我们需要创建它们，这就要用到消息会话工厂类——MessageSessionFactory，由这个工厂帮你创建生产者或者消费者。除了这些，MessageSessionFactory还在后面帮你做很多事情，包括：

• 服务的查找和发现，通过Diamond和Zookeeper帮你查找日常的Meta服务器地址列表。
• 连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台Meta的服务器在一个工厂内只维持一个连接。
• 消息消费者的消息存储和恢复。
• 协调和管理各种资源，包括创建的生产者和消费者的。

消息生产者的接口是MessageProducer，你可以通过它来发送消息。创建生产者很简单，通过MessageSessionFactory的createProducer方法即可以创建一个生产者。请注意，MessageProducer是线程安全的，完全可重复使用，因此最好在应用中作为单例来使用，一次创建，到处使用，配置为Spring里的singleton bean。MessageProducer创建的代价昂贵，每次都需要通过zk查找服务器并创建TCP长连接。

发送消息后，消费者可以接收消息了。通过createConsumer方法来创建MessageConsumer，传入一个ConsumerConfig参数，这是消费者的配置对象。每个消息者都必须有一个ConsumerConfig配置对象，这里只设置了group属性，这是消费者的分组名称。Meta的Producer、Consumer和Broker都可以为集群。消费者可以组成一个集群共同消费同一个Topic，发往这个Topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。同一个消费者集群必须拥有同一个分组名称，也就是同一个Group，这个概念跟Notify里的订阅者组名是一样的。消息的消费过程可以是一个并发处理的过程，getExecutor返回你想设置的线程池，每次消费都会在这个线程池里进行。recieveMessage方法用于实际的消息消费处理，message参数即为消费者收到的消息，它必不为空。

如果在消费过程中抛出任何异常，该条消息将会在一定间隔后重新尝试提交给MessageListener消费。在多次消费失败的情况下，该消息将会存储到消费者应用的本次磁盘，并在后台自动恢复重试消费。

Metamorphosis 1.2开始支持事务，包括发送端和消费端事务。发送端同时支持本地事务和分布式事务，可以在一个事务内发送多条消息，要么同时成功，要么同时失败；可以使用XA事务，在事务内操作其他XA资源，例如操作数据库，与此同时发送meta消息，可以保证这些操作和发送消息要么一起成功，要么一起失败。

如果你要在发送消息的同时操作数据库，比如同时将消息插入某张表，例如下订单的时候同时发送消息通知卖家并将订单插入数据库，这时候因为涉及到两个Resource（数据库和Meta)，就需要使用分布式事务来保证ACID。分布式事务一般采用两阶段提交协议，在java里就是使用JTA规范API的XA部分。

关于服务器的可靠性保证，消息生产者发送的消息在Meta服务器收到后在做必要的校验和检查之后的第一件事就是写入磁盘，写入成功之后返回应答给生产者。因此，可以确认每条发送结果为成功的消息服务器都是写入磁盘的。

写入磁盘，不意味着数据落到磁盘设备上，毕竟我们还隔着一层OS，OS对写有缓冲。Meta有两个特性来保证数据落到磁盘上：

1. 每1000条（可配置），即强制调用一次force来写入磁盘设备。
2. 每隔10秒（可配置），强制调用一次force来写入磁盘设备。

因此，Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。

服务器通常组织为一个集群，一条从生产者过来的消息可能按照路由规则存储到集群中的某台机器。Meta还正在实现高可用的HA方案，类似MySQL的异步复制，将一台Meta服务器的数据完整复制到另一台Slave服务器，并且Slave服务器还提供消费功能，本方案正在实现中。

对于消费者的可靠性保证 ，消息的消费者是一条接着一条地消费消息，只有在成功消费一条消息后才会接着消费下一条。如果在消费某条消息失败（如异常），则会尝试重试消费这条消息（默认最大5次），超过最大次数后仍然无法消费，则将消息存储在消费者的本地磁盘，由后台线程继续做重试。而主线程继续往后走，消费后续的消息。因此，只有在MessageListener确认成功消费一条消息后，meta的消费者才会继续消费另一条消息。由此来保证消息的可靠消费。

消费者的另一个可靠性的关键点是Offset的存储，也就是拉取数据的偏移量。我们目前提供了以下几种存储方案：

1. Zookeeper，默认存储在Zoopkeeper上，Zookeeper通过集群来保证数据的安全性。
2. MySQL，可以连接到您使用的MySQL数据库，只要建立一张特定的表来存储。完全由数据库来保证数据的可靠性。
3. File，文件存储，将Offset信息存储在消费者的本地文件中。

Offset会定期保存，并且在每次重新负载均衡前都会强制保存一次。

Meta的HA(High Availability)提供了在某些Broker出现故障时继续工作而不影响消息服务的可用性；跟HA关系紧密的就是Failover，当故障Server恢复时能重新加入Cluster处理请求，这个过程对消息服务的使用者是透明的。Meta基于Master/Slave实现HA，Slave以作为Master的订阅者（consumer）来跟踪消息记录，当消息发送到Master时候，Slave会定时的获取此消息记录，并存储在自己的Store实现上；当Master出现故障无法继续使用了，消息还会在Slave上Backup的记录。这种方式不影响原有的消息的记录，一旦Master记录成功，就返回成功，不用等待在slave上是否记录；正因如此，Slave和Master还有稍微一点的时间差异，在Master出故障那一瞬间，或许有最新产生的消息，就无法同步到Slave；另外Slave可以作为Consumer的服务提供者，意思就是如果要写入必须通过Master，消费时候可以从Slave上直接获取。

Failover机制采用Client端方式，Master和Slave都需要注册到ZK上，一旦Master无法使用，客户端可使用与之对应的Slave；当Master的故障恢复时候，处理的方式：原来的Master变成Slave，Slave变成Master；恢复故障的broker作为Slave去之前的Slave同步消息。优点简单，但是需要Slave和Master有一样的配置和处理能力，这样就能取代Master的位置，目前Meta采用此方式。

对Metamorphosis感兴趣的开发人员可以访问其 官方网站，了解更多信息。

崔康 热情的技术探索者，资深软件工程师，InfoQ编辑，从事企业级Web应用的相关工作，关注性能优化、Web技术、浏览器等领域。