Metamorphosis分布式消息中间件

一 简介

1.1定义

    Metamorphosis是淘宝开源的一个Java消息中间件。关于消息中间件，你应该听说过JMS（1）规范，以及一些开源实现，如ActiveMQ和HornetQ等。Metamorphosis也是其中之一。

    Metamorphosis是一款完全的队列模型消息中间件，服务器使用Java语言编写，可在多种软硬件平台上部署。客户端支持Java、C++编程语言。单台服务器可支持1万以上个消息队列，通过扩容服务器，队列数几乎可任意横向扩展。每个队列都是持久化、长度无限（取决于磁盘空间大小）、并且可从队列任意位置开始消费。

    Metamorphosis (MetaQ) 是一个高性能、高可用、可扩展的分布式消息中间件，类似于LinkedIn的Kafka，具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，在淘宝和支付宝有着广泛的应用，现已开源。

1.2起源

       Metamorphosis的起源是linkedin的开源MQ，这是一个设计很独特的MQ系统，它采用pull机制，而不是一般MQ的push模型，它大量利用了zookeeper做服务发现和offset存储，总体上说Metamorphosis的设计跟它是完全一致的。但是为什么还需要meta呢？

       简单概括下重新写出meta的原因：

·       Kafka是scala写，作者对scala不熟悉，并且kafka整个社区的发展太缓慢了。

·       有一些功能是kakfa没有实现，但是我们却需要：事务、多种offset存储、高可用方案(HA)等。

       Meta相对于kafka特有的一些功能：

·       文本协议设计，非常透明，支持类似memcached stats的协议来监控broker

·       纯Java实现，从通讯到存储，从client到server都是重新实现。

·       提供事务支持，包括本地事务和XA（2）分布式事务。

·       支持HA（High Availability）复制，包括异步复制和同步复制，保证消息的可靠性。

·       支持异步发送消息。

·       消费消息失败，支持本地恢复。

·       多种offset存储支持，数据库、磁盘、zookeeper，可自定义实现。

·       支持group commit，提升数据可靠性和吞吐量。

·       支持消息广播模式。

·       一系列配套项目：python客户端、twitter storm的spout、tail4j等。

 

1.3内部机制与总体架构

       从实现角度看，MetaQ充分利用zookeeper这个优秀的服务中心，作为服务注册和查找中心、客户端负载均衡和数据偏移量的分布式存储使用。Zookeeper在MetaQ整个集群中扮演非常关键的角色。

       MetaQ的存储实现与kafka是一致的，充分利用传统磁盘顺序读写非常高效的特点，并且利用group commit、sendfile系统调用等技术来充分提高IO效率。

       MetaQ的事务实现跟ActiveMQ是类似的，采用redo日志的方式，并遵循JTA（4）协议规范来实现分布式事务。

       MetaQ的网络协议跟memcached文本协议类似，但MetaQ的协议引入了opaque的映射字段，提高请求的并行度。正因为采用文本协议，为MetaQ编写其他语言客户端是相对容易，并且管理MetaQ服务器也变的很容易，比如MetaQ提供了stats协议，通过telent就可以获取服务器的运行状况。

       下图给出了它的总体架构：

1.4应用

    因此meta相比于kafka的提升是巨大的。meta在淘宝和支付宝都得到了广泛应用，现在每天支付宝每天经由meta路由的消息达到120亿，淘宝也有每天也有上亿的消息量。总计METAQ在阿里巴巴各个子公司被广泛应用，每天会转发250亿+条消息。下面是使用METAQ的各个公司或项目名：淘宝网，支付宝，阿里巴巴金融，阿里云，Alibaba.com,Alipay.com.

 

 

       据说,腾讯和京东也在尝试使用。

       Meta适合的应用：

·       日志传输，高吞吐量的日志传输本来就是kafka的强项

·       消息广播功能，如广播缓存配置失效。

·       数据的顺序同步功能，如mysql binlog复制

·       分布式环境下（broker,producer,consumer都为集群）的消息路由，对顺序和可靠性有极高要求的场景。

·       作为一般MQ来使用的其他功能

注释：

1 JMS（Java Messaging Service）是Java平台上有关面向消息中间件(MOM)的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发，翻译为Java消息服务。

2  XA 就是 X/Open DTP 定义的交易中间件与数据库之间的接口规范（即接口函数），交易中间件用它来通知数据库事务的开始、结束以及提交、回滚等。 XA 接口函数由数据库厂商提供。 

3 Diamond是淘宝网Java中间件团队的核心产品之一，服务于集团线上很多核心应用。目前已经开源，Diamond主要针对的是持久数据，可以和ZookKeeper类比。

4 JTA，即Java Transaction API，译为Java事务API。

二 可靠性、顺序和重复

2.1可靠性

       Metamorphosis的可靠性保证贯穿客户端和服务器。

2.1.1生产者的可靠性保证

消息生产者发送消息后返回SendResult，如果isSuccess返回为true,则表示消息已经确认发送到服务器并被服务器接收存储。整个发送过程是一个同步的过程。保证消息送达服务器并返回结果。

2.1.2服务器的可靠性保证

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

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

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

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

因此,Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。当然通过参数调整你甚至可以在掉电情况下不丢失任何消息。

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

2.1.3消费者的可靠性保证

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

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

1 zookeeper，默认存储在zoopkeeper上，zookeeper通过集群来保证数据的安全性。

2 mysql，可以连接到您使用的mysql数据库，只要建立一张特定的表来存储。完全由数据库来保证数据的可靠性。

3 file，文件存储，将offset信息存储在消费者的本地文件中。

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

2.2顺序

很多人关心的消息顺序，希望消费者消费消息的顺序跟消息的发送顺序是一致的。比如，我发送消息的顺序是A、B、C，那么消费者消费的顺序也应该是A、B、C。乱序对某些应用可能是无法接受的。

Metamorphosis对消息顺序性的保证是有限制的，默认情况下，消息的顺序以谁先达到服务器并写入磁盘，则谁就在先的原则处理。并且，发往同一个分区的消息保证按照写入磁盘的顺序让消费者消费，这是因为消费者针对每个分区都是按照从前到后递增offset的顺序拉取消息。

Meta可以保证，在单线程内使用该producer发送的消息按照发送的顺序达到服务器并存储，并按照相同顺序被消费者消费，前提是这些消息发往同一台服务器的同一个分区。为了实现这一点，你还需要实现自己的PartitionSelector用于固定选择分区

public interface PartitionSelector {

    public Partition getPartition(String topic, Listpartitions, Message message) throws MetaClientException;

}

选择分区可以按照一定的业务逻辑来选择，如根据业务id来取模。或者如果是传输文件，可以固定选择第n个分区使用。当然，如果传输文件，通常建议只配置一个分区，那也就无需选择了。

2.3消息重复

消息的重复包含两个方面，生产者重复发送消息以及消费者重复消费消息。

针对生产者来说，有可能发生这种情况，生产者发送消息，等待服务器应答，这个时候发生网络故障，服务器实际已经将消息写入成功，但是由于网络故障没有返回应答。那么生产者会认为发送失败，则再次发送同一条消息，如果发送成功，则服务器实际存储两条相同的消息。这种由故障引起的重复，meta是无法避免的，因为meta不判断消息的data是否一致，因为它并不理解data的语义，而仅仅是作为载荷来传输。

针对消费者来说也有这个问题，消费者成功消费一条消息，但是此时断电，没有及时将前进后的offset存储起来，则下次启动的时候或者其他同个分组的消费者owner到这个分区的时候，会重复消费该条消息。这种情况meta也无法完全避免。

Meta对消息重复的保证只能说在正常情况下保证不重复，异常情况无法保证，这些限制是由远程调用的语义引起的，要做到完全不重复的代价很高，meta暂时不会考虑。

三 如何开始

下载服务器

从下载页面选择最新版本的服务器(目前是1.4.4)并下载到本地，假设下载后的文件在folder目录，执行下列命令解压缩文件：

cd folder

tar zxvf   taobao-metamorphosis-server-wrapper-1.4.4.tar.gz

解压缩文件,解压后目录结构大概为：

taobao

    metamorphosis-server-wrapper

         bin

            env.bat               

            env.sh                

            log4j.properties      

            metaServer.bat        

            metaServer.sh         

            tools_log4j.properties

         logs

         conf

            server.ini

            ......

         lib

            ......

启动脚本放在bin目录，主要的脚本是metaServer.sh，日志在logs目录，而配置文件主要是conf目录下server.ini，lib存放所有的依赖jar包。

配置Broker

默认server.ini提供了一个topic——test用于测试，你可以添加自己定义的topic:

[topic=mytopic]                                                                                                                                                                              

;是否启用统计                                                                                                                                                                            

stat=true                                                                                                                                                                                   

;这个topic指定分区数目，如果没有设置，则使用系统设置                                                                                                                                         

numPartitions=10                                                                                                                                                                                                                                                                                                                                                    

;删除策略的执行时间，cron表达式                                                                                                                                                               

deleteWhen=0 0 6,18 * * ?

大多数system参数都可以topic参数所覆盖。

启动和关闭服务器

确保你的机器上安装了JDK并正确设置JAVA_HOME和PATH变量，启动服务器：

bin/metaServer.sh start local

关闭服务器：

bin/metaServer.sh stop

在windows上，双击执行"bin/metaServer.bat"即可(windows不支持local模式启动，需配置zookeeper，见下文的zookeeper配置一节）。

修改JAVA_HOME,JMX等变量，请修改bin/env.sh(for linux)或者bin/env.bat(for windows)。

更多metaServer.sh支持命令请使用help:

 bin/metaServer.sh help

=>

  Usage: metaServer.sh {start|status|stop|restart|reload|stats|open-partitions|close-partitions|move-partitions|delete-partitions|query}

  ......

验证服务器正常运行

除了通过观察日志logs/metaServer.log外，你还可以通过stats命令来观察服务器运行状况。

 bin/metaServer.sh stats

=>

   STATS

   pid 7244

   broker_id 0

   port 8123

   uptime 2057

   version 1.4.2

   curr_connections 1

   threads 35

   cmd_put 0

   cmd_get 0

   cmd_offset 0

   tx_begin 0

   tx_xa_begin 0

   tx_commit 0

   tx_rollback 0

   get_miss 0

   put_failed 0

   total_messages 0

   topics 2

   config_checksum 718659887

   END

你也可以telnet到默认的8123端口执行stats命令查看。

集群模式配置

Local模式

上文提到的启动方式是以本地模式也就是单机模式启动，它将启动一个内置的zookeeper，并将broker注册到该zookeeper。这对于单机应用或者测试开发是最便捷方式的。

但是MetaQ是作为分布式软件设计的，更通常作为一个集群提供服务。MetaQ的集群管理是利用zookeeper实现的，因此首先需要配置zookeeper。