第一部分    MetaQ简介

1.简介

MetaQ（全称Metamorphosis）是一个高性能、高可用、可扩展的分布式消息中间件，思路起源于LinkedIn的Kafka，但并不是Kafka的一个Copy。MetaQ具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，目前在淘宝和支付宝有着广泛的应用.

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

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

• 文本协议设计，非常透明，支持类似memcached stats的协议来监控broker
• 纯Java实现，从通讯到存储，从client到server都是重新实现。
• 提供事务支持，包括本地事务和XA分布式事务
• 支持HA复制，包括异步复制和同步复制，保证消息的可靠性
• 支持异步发送消息
• 消费消息失败，支持本地恢复
• 多种offset存储支持，数据库、磁盘、zookeeper，可自定义实现
• 支持group commit，提升数据可靠性和吞吐量。（目前kafka已实现）
• 支持消息广播模式
• 一系列配套项目：Python/Ruby/C/C++客户端、Twitter Storm的Spout、Tail4j等。

ActiveMQ和HornetQ都是符合Java EE中JMS规范的MQ实现，两者都是很优秀的开源MQ。同时两者也不局限在JMS规范，同时也支持其他一些MQ协议，如stomp协议、AMQP协议等。相比来说，就我当时了解的情况来看，HornetQ的性能会比ActiveMQ更强，因为HornetQ使用JNI基于异步IO做了更多优化，而对于MQ来说，最终的瓶颈都是落在IO存储上。MetaQ的性能是远远超过这两个MQ的，有一个网友做的比较可以说明一定问题（http://www.blogjava.net/livery/articles/391595.html），但是这不能说MetaQ比它们就更优秀，因为这跟它们的实现，和面对的场景有很大关系。
    ActiveMQ和HornetQ，这两个MQ从诞生起就是为了企业应用而设计的，JMS规范本身也是企业应用系统的规范。这一套东西，我个人认为并不适合互联网应用。互联网应用通常面对的是海量的数据，并且通常对事务一致性的要求相对较弱，而企业应用对事务一致性的要求就相对很高。互联网应用为了处理大量请求，通常采用集群处理的方式，而JMS规范并不重视分布式应用。我说的这个集群不仅仅是服务端broker的集群，还包括生产者和消费者都可能是一个又一个集群，而传统的JMS规范是没有明确处理这些情况的。互联网应用还有一个问题是异构系统特别多，而JMS规范只是Java EE这个平台上的规范，对异构系统的接入也是一个比较麻烦的地方，不同的实现有很大的差异。

综合来讲，HornetQ和ActiveMQ是为了企业级应用设计的消息中间件，而MetaQ从一开始就是为了大规模互联网应用设计的消息中间件，两者面对的场景和需求不同。开发者可根据实际的需求，选择合适的产品。

从MQ的发展来看，我们可以看到，出现了越来越多特定领域的消息中间件，例如memcacheq、kestrel、beanstalkd甚至redis，它们很轻量级，并且不想做到全能，而只是解决一个领域的问题，我觉得这是未来的趋势。

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

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

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

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

• 日志传输，高吞吐量的日志传输，这本来也是kafka的强项。
• 消息广播功能，如广播缓存配置失效。
• 数据的顺序同步功能，如MySQL binlog复制。
• 分布式环境下（broker、producer、consumer都为集群）的消息路由，对顺序和可靠性有极高要求的场景。
• 作为一般MQ来使用的其他功能。

MetaQ作为一个分布式的消息中间件，需要依赖zookeeper，对于一些规模不大、单机应用的场景，我个人并不是特别支持尝试用MetaQ，因为多一个依赖系统，其实就是多一份风险，在这些简单场景下，可能类似memcacheq、kestrel甚至redis等轻量级MQ就非常合适。而MetaQ一开始就是为大规模分布式系统设计的，如果不当使用，可能没有带来好处，反而多出一堆问题。开发者需要根据自己面对的场景，团队的技术能力，做出一个合适的选择。

第二部分   实例

1.概念和术语

消息生产者

也称为Message Producer，一般简称为producer，负责产生消息并发送消息到meta服务器。

消息消费者

也称为Message Consumer，一般简称为consumer，负责消息的消费，meta采用pull模型，由消费者主动从meta服务器拉取数据并解析成消息并消费。

Topic

消息的主题，由用户定义并在服务端配置。producer发送消息到某个topic下，consumer从某个topic下消费消息。

分区(partition)

同一个topic下面还分为多个分区，如meta-test这个topic我们可以分为10个分区，分别有两台服务器提供，那么可能每台服务器提供5个分区，假设服务器id分别为0和1，则所有分区为0-0、0-1、0-2、0-3、0-4、1-0、1-1、1-2、1-3、1-4。

分区跟消费者的负载均衡机制有很大关系，具体见集群和负载均衡。

Message

消息，负载用户数据并在生产者、服务端和消费者之间传输。

Broker

就是meta的服务端或者说服务器，在消息中间件中也通常称为broker。

消费者分组(Group)

消费者可以是多个消费者共同消费一个topic下的消息，每个消费者消费部分消息。这些消费者就组成一个分组，拥有同一个分组名称,通常也称为消费者集群

Offset

消息在broker上的每个分区都是组织成一个文件列表，消费者拉取数据需要知道数据在文件中的偏移量，这个偏移量就是所谓offset。Offset是绝对偏移量，服务器会将offset转化为具体文件的相对偏移量。详细内容参见#消息的存储结构

2.可靠性、顺序和重复

可靠性

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

生产者的可靠性保证

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

服务器的可靠性保证

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

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

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

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

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

消费者的可靠性保证

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

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

• zookeeper，默认存储在zoopkeeper上，zookeeper通过集群来保证数据的安全性。
• mysql，可以连接到您使用的mysql数据库，只要建立一张特定的表来存储。完全由数据库来保证数据的可靠性。
• file，文件存储，将offset信息存储在消费者的本地文件中。

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

顺序

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

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

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

public interface PartitionSelector {

public Partition getPartition(String topic, List<Partition> partitions, Message message) throws MetaClientException;

}

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

消息的顺序发送我们在1.2这个版本提供了OrderedMessageProducer，自定义管理分区信息，并提供故障情况下的本地存储功能。

消息重复

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

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

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

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

3.如何开始

如何开始

下载服务器

从下载页面选择最新版本的服务器(目前是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。 一个MetaQ集群必须使用同一个zookeeper集群。

配置zookeeper

Metamorphosis使用zookeeper发布和订阅服务，并默认使用zookeeper存储消费者offset，因此，你需要首先安装一个zookeeper到某台机器上，或者使用某个现有的zk集群，安装zookeeper请参考zookeeper文档。

假设你在本机安装并启动了一个zookeeper，端口在默认的2181，请修改conf/server.ini文件，保证zookeeper地址正确：

;zk的服务器列表

zk.zkConnect=localhost:2181

;zk心跳超时，单位毫秒，默认30秒

zk.zkSessionTimeoutMs=30000

;zk连接超时时间，单位毫秒，默认30秒

zk.zkConnectionTimeoutMs=30000

;zk数据同步时间，单位毫秒，默认5秒

zk.zkSyncTimeMs=5000

其他zk参数请酌情设置。其他重要参数还包括dataPath和numPartitions，分别用于指定默认的数据存储路径，以及默认topic的分区数目。具体参数信息请看conf/server_sample.ini示范文件。

启动服务器

假设你准备建立一个两台broker组成的MetaQ集群提供消息服务，那么通常你需要两台机器来运行MetaQ。但是这里为了简单测试，以单机运行集群模式为例子。

首先，按照上文的提示，正常启用Local模式后，假设为broker1，请拷贝一份MetaQ到不同目录，假设为broker2，需要修改两个参数：serverPort和dataPath，分别对应服务器端口和数据存储路径，将broker2的这两个参数修改为不同值，其他设置保持不变。

然后，停止local模式启动的broker1， 并重新以集群模式启动：

metaServer.sh stop

metaServer.sh start

进入broker2的bin目录，启动broker2:

metaServer.sh start

这样即组成两个broker的MetaQ集群。

4.简单例子

简单例子

示例源码

• metamorphosis-example

Example

消息中间件中有两个角色：消息生产者和消息消费者。Meta里同样有这两个概念，消息生产者负责创建消息并发送到meta服务器，meta服务器会将消息持久化到磁盘，消息消费者从meta服务器拉取消息并提交给应用消费。我们假设你已经部署了你的meta服务器，参见如何开始。

Java客户端例子

推荐你使用maven，引用meta java client非常简单：

<dependency>

<groupId>com.taobao.metamorphosis</groupId>

<artifactId>metamorphosis-client</artifactId>

<version>1.4.4</version>

</dependency>

创建一个示例工程，或者直接将metamorphosis-example clone出来到你本地机器。

git clone git://github.com/killme2008/metamorphosis-example.git

请注意，1.4.3及以上版本的java客户端只能连接1.4.3及以上版本的MetaQ服务器，而1.4.3之前的老客户端则没有限制。主要是因为1.4.3引入了发布和订阅topic的分离，1.4.3的新客户端只能查找到新版本的broker

消息会话工厂类

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

1.服务的查找和发现，通过diamond和zookeeper帮你查找日常的meta服务器地址列表

2.连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台meta的服务器在一个工厂内只维持一个连接。

3.消息消费者的消息存储和恢复，后续我们会谈到这一点。

4.协调和管理各种资源，包括创建的生产者和消费者的。

因此，我们首先需要创建一个会话工厂类，MessageSessionFactory仅是一个接口，它的实现类是MetaMessageSessionFactory：

MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(new MetaClientConfig());

请注意,MessageSessionFactory应当尽量复用，也就是作为应用中的单例来使用，简单的做法是交给spring之类的容器帮你托管。

消息生产者

package com.taobao.metamorphosis.example;

import java.io.BufferedReader;

import java.io.InputStreamReader;

import com.taobao.metamorphosis.Message;

import com.taobao.metamorphosis.client.MessageSessionFactory;

import com.taobao.metamorphosis.client.MetaClientConfig;

import com.taobao.metamorphosis.client.MetaMessageSessionFactory;

import com.taobao.metamorphosis.client.producer.MessageProducer;

import com.taobao.metamorphosis.client.producer.SendResult;

import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig;

public class Producer {

public static void main(String[] args) throws Exception {

final MetaClientConfig metaClientConfig = new MetaClientConfig();

final ZKConfig zkConfig = new ZKConfig();

//设置zookeeper地址

zkConfig.zkConnect = "127.0.0.1:2181";

metaClientConfig.setZkConfig(zkConfig);

// New session factory,强烈建议使用单例

MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);

// create producer,强烈建议使用单例

MessageProducer producer = sessionFactory.createProducer();

// publish topic

final String topic = "test";

producer.publish(topic);

BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

String line = null;

while ((line = reader.readLine()) != null) {

// send message

SendResult sendResult = producer.sendMessage(new Message(topic, line.getBytes()));

// check result

if (!sendResult.isSuccess()) {

System.err.println("Send message failed,error message:" + sendResult.getErrorMessage());

}

else {

System.out.println("Send message successfully,sent to " + sendResult.getPartition());

}

}

}

}

消息生产者的接口是MessageProducer，你可以通过它来发送消息。创建生产者很简单，通过MessageSessionFactory的createProducer方法即可以创建一个生产者。在Meta里，每个消息对象都是Message类的实例，Message表示一个消息对象，它包含这么几个属性：

• id: Long型的消息id,消息的唯一id，系统自动产生，用户无法设置，在发送成功后由服务器返回，发送失败则为0。
• topic: 消息的主题，订阅者订阅该主题即可接收发送到该主题下的消息，生产者通过指定发布的topic查找到需要连接的服务器地址，必须。
• data: 消息的有效载荷，二进制数据，也就是消息内容，meta永远不会修改消息内容，你发送出去是什么样子，接收到就是什么样子。消息内容通常限制在1M以内，我的建议是最好不要发送超过上百K的消息，必须。数据是否压缩也完全取决于用户。
• attribute: 消息属性，一个字符串，可选。发送者可设置消息属性来让消费者过滤。

细心的朋友可能注意到，我们在sendMessage之前还调用了MessageProducer的publish(topic)方法

producer.publish(topic);

这一步在发送消息前是必须的，你必须发布你将要发送消息的topic，这是为了让会话工厂帮你去查找接收这些topic的meta服务器地址并初始化连接。这个步骤针对每个topic只需要做一次，多次调用无影响。

总结下这个例子，从标准输入读入你输入的数据，并将数据封装成一个Message对象，发送到topic为test的服务器上。

请注意，MessageProducer是线程安全的，完全可重复使用，因此最好在应用中作为单例来使用，一次创建，到处使用，配置为spring里的singleton bean。MessageProducer创建的代价昂贵，每次都需要通过zk查找服务器并创建tcp长连接。

消息消费者

发送消息后，消费者可以接收消息了，下面的代码创建消费者并订阅test这个主题，等待消息送达并打印消息内容

package com.taobao.metamorphosis.example;

import java.util.concurrent.Executor;

import com.taobao.metamorphosis.Message;

import com.taobao.metamorphosis.client.MessageSessionFactory;

import com.taobao.metamorphosis.client.MetaClientConfig;

import com.taobao.metamorphosis.client.MetaMessageSessionFactory;

import com.taobao.metamorphosis.client.consumer.ConsumerConfig;

import com.taobao.metamorphosis.client.consumer.MessageConsumer;

import com.taobao.metamorphosis.client.consumer.MessageListener;

import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig;

public class AsyncConsumer {

public static void main(String[] args) throws Exception {

final MetaClientConfig metaClientConfig = new MetaClientConfig();

final ZKConfig zkConfig = new ZKConfig();

//设置zookeeper地址

zkConfig.zkConnect = "127.0.0.1:2181";

metaClientConfig.setZkConfig(zkConfig);

// New session factory,强烈建议使用单例

MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);

// subscribed topic

final String topic = "test";

// consumer group

final String group = "meta-example";

// create consumer,强烈建议使用单例

MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfig(group));

// subscribe topic

consumer.subscribe(topic, 1024 * 1024, new MessageListener() {

public void recieveMessages(Message message) {

System.out.println("Receive message " + new String(message.getData()));

}

public Executor getExecutor() {

// Thread pool to process messages,maybe null.

return null;

}

});

// complete subscribe

consumer.completeSubscribe();

}

}

通过createConsumer方法来创建MessageConsumer，注意到我们传入一个ConsumerConfig参数，这是消费者的配置对象。每个消息者都必须有一个ConsumerConfig配置对象，我们这里只设置了group属性，这是消费者的分组名称。Meta的Producer、Consumer和Broker都可以为集群。消费者可以组成一个集群共同消费同一个topic，发往这个topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。同一个消费者集群必须拥有同一个分组名称，也就是同一个group。我们这里将分组名称设置为meta-example。

订阅消息通过subscribe方法，这个方法接受三个参数

• topic，订阅的主题
• maxSize，因为meta是一个消费者主动拉取的模型，这个参数规定每次拉取的最大数据量，单位为字节，这里设置为1M，默认最大为1M。
• MessageListener，消息监听器，负责消息消息。

MessageListener的接口方法如下:

public interface MessageListener {

/**

* 接收到消息列表，只有message不为空并且不为null的情况下会触发此方法

*

* @param message

*/

public void recieveMessages(Message message);

/**

* 处理消息的线程池

*

* @return

*/

public Executor getExecutor();

}

消息的消费过程可以是一个并发处理的过程，getExecutor返回你想设置的线程池，每次消费都会在这个线程池里进行。recieveMessage方法用于实际的消息消费处理，message参数即为消费者收到的消息，它必不为null。

我们这里简单地打印收到的消息内容就完成消费。如果在消费过程中抛出任何异常，该条消息将会在一定间隔后重新尝试提交给MessageListener消费。在多次消费失败的情况下，该消息将会存储到消费者应用的本次磁盘，并在后台自动恢复重试消费。

细心的你一定还注意到，在调用subscribe之后，我们还调用了completeSubscribe方法来完成订阅过程。请注意，subscribe仅是将订阅信息保存在本地，并没有实际跟meta服务器交互，要使得订阅关系生效必须调用一次completeSubscribe，completeSubscribe仅能被调用一次，多次调用将抛出异常。 为什么需要completeSubscribe方法呢，原因有二：

• 首先，subscribe方法可以被调用多次，也就是一个消费者可以消费多种topic
• 其次，如果每次调用subscribe都跟zk和meta服务器交互一次，代价太高

因此completeSubscribe一次性将所有订阅的topic生效，并处理跟zk和meta服务器交互的所有过程。

同样,MessageConsumer也是线程安全的，创建的代价不低，因此也应该尽量复用。

5.配置管理

示例配置

完整的参数配置示例：

• sample.server.ini

服务端配置

Meta服务端配置主要在服务器conf目录下的server.ini文件，整体配置分为三部分：系统参数、zookeeper参数以及topic配置。系统参数在system section，zookeeper参数配置在zookeeper section，而topic的配置是在topic=xxxx section。具体说明如下：

一份默认提供的参数配置在这里。

系统参数部分

系统参数配置都放在[system]下面：

• brokerId: 服务器集群中唯一的id，必须为整型0-1024之间。对服务器集群的定义是使用同一个zookeeper并且在zookeeper上的root path相同，具体参见zookeeper配置。

hostName: 服务器hostname，默认取本机IP地址，如果你是多网卡机器，可能需要明确指定。服务器会将此hostname加上端口写入到zookeeper提供给客户端发现。

serverPort:服务器端口，默认8123。PS. 选择8123是因为这蕴含着我儿子的生日 :D。

numPartitions:系统默认情况下每个topic的分区数目，默认为1，可被topic配置覆盖。单个服务器的总分区数目不建议超过1000，太多将导致频繁的磁盘寻道严重影响IO性能。

dataPath: 服务器数据文件路径，默认在~home/meta下,每个topic可以覆盖此配置，对于多块磁盘的机器，可设置不同topic到不同磁盘来提升IO效率。

dataLogPath：数据日志文件路径，主要存放事务日志，默认跟dataPath一致，最好单独设置到不同的磁盘或者目录上。如果为空，使用指定的dataPath

getProcessThreadCount: 处理get请求的并发线程数，默认为CPUS*10。

putProcessThreadCount: 处理put请求的并发线程数，默认为CPUS*10。

maxSegmentSize: 单个数据文件的大小，默认为1G。默认无需修改此选项。

maxTransferSize: 传输给消费者的最大数据大小，默认为1M，请根据你的最大消息大小酌情设置，如果太小，每次无法传输一个完整的消息给消费者，导致消费者消费停滞。可设置成一个大数来取消限制。

1.4.3版本引入的参数：

acceptPublish: 是否接收消息，默认为true；如果为false，则不会注册发送信息到zookeeper上，客户端当然无法发送消息到该broker。本参数可以被后续的topic配置覆盖。

acceptSubscribe: 与acceptPublish类似，默认也为true；如果为false，则不会注册消费信息到zookeeper上，消费者无法发现该broker，当然无法从该broker消费消息。本参数可以被后续的topic配置覆盖。

1.4.4版本新引入参数：

• stat：全局性地控制是否开启实时统计，可被topic配置覆盖，默认为false。
• loadMessageStoresInParallel: 是否启动时并行加载数据，开启可提升启动速度。默认不开启。开启后启动日志顺序可能紊乱。
• updateConsumerOffsets: 当消费者的offset不在Broker的数据范围内，是否强制更新消费者的offset为当前最大offset。默认为false。测试开发环境建议开启此选项，生产环境不建议。

数据可靠性参数

Meta保证消息可靠性是建立在磁盘可靠性的基础上，发送的每一条消息都保证是在“写入磁盘”的情况下才返回给客户端应答。这里有两个关键参数可以控制：

• unflushThreshold: 每隔多少条消息做一次磁盘sync，强制将更改的数据刷入磁盘。默认为1000。也就是说在掉电情况下，最多允许丢失1000条消息。可设置为0，强制每次写入都sync。在设置为0的情况下，服务器会自动启用group commit技术，将多个消息合并成一次sync来提升IO性能。经过测试，group commit情况下消息发送者的TPS没有受到太大影响，但是服务端的负载会上升很多。
• unflushInterval: 间隔多少毫秒定期做一次磁盘sync,默认是10秒。也就是说在服务器掉电情况下，最多丢失10秒内发送过来的消息。不可设置为小于或者等于0。

请注意，上述两个参数都可以被topic单独配置说覆盖，也就是说每个topic可以配置不同的数据可靠级别。

当某个topic开启group commit后，将为每个分区配置一个线程做聚集force，因此请控制启用group commit技术的topic数量，太多可能导致过多线程，反而效率下降。

数据删除策略配置

默认情况下，meta是会保存不断添加的消息，然后定期对“过期”的数据进行删除或者归档处理，这都是通过下列参数控制的：

• deleteWhen: 何时执行删除策略的cron表达式，默认是0 0 6,18 * * ?，也就是每天的早晚6点执行处理策略。
• deletePolicy: 数据删除策略，默认超过7天即删除,这里的168是小时，10s表示10秒，10m表示10分钟，10h表示10小时，不明确指定单位默认为小时。delete是指删除，超过指定时间的数据文件将被彻底从磁盘删除。也可以选择archive策略，即不对过期的数据文件做删除而是归档，当使用archive策略的时候可以选择是否压缩数据文件，如167,archive,true即选择将更改时间超过7天的数据文件归档并压缩为zip文件，如果不选择压缩，则重命名为扩展名为arc的文件。

上述两个参数都可以被topic单独配置所覆盖，也就是每个topic可以指定自己独特的删除策略。通常来说，对于不重要的topic可以将更早地将他们删除来节省磁盘空间。

事务相关配置

• maxCheckpoints: 最大保存事务checkpoint数目，默认为3，服务器在启动的时候会从最近一次checkpoint回访事务日志文件，恢复重启前的事务状态。不建议修改此参数。
• checkpointInterval：事务checkpoint时间间隔，单位毫秒，默认1小时。间隔时间太长，会导致启动的时候replay事务日志占用了太多时间，太短则可能影响到性能。
• maxTxTimeoutTimerCapacity：最大事务超时timer的数量。服务端会为每个事务启动一个定时器监控事务是否超时，定时器的数目上限通过本参数限制。限制了本参数，也变相地控制了最大可运行的事务数。默认为30000个。
• maxTxTimeoutInSeconds：最大事务超时时间，单位为秒，默认为60秒。客户端设置的事务超时时间不能超过此设定，超过将被强制限制为此设定。
• flushTxLogAtCommit：服务端对事务日志的sync策略，0表示让操作系统决定，1表示每次commit都刷盘，2表示每隔1秒刷盘一次。此参数严重影响事务性能，可根据你需要的性能和可靠性之间权衡做出一个合理的选择。通常建议设置为2，表示每隔1秒刷盘一次，也就是最多丢失一秒内的运行时事务。这样的可靠级别对大多数服务是足够的。最安全的当然是设置为1，但是将严重影响事务性能。而0的安全级别最低。安全级别上 1>=2>0，而性能则是0 >= 2 > 1。

zookeeper配置

meta服务端会将自身id,topic信息和socket地址发送到zookeeper上，让客户端可以发现并连接服务器。Zookeeper相关的配置放在[zookeeper]模块下面:

• zk.zkEnable: 是否启用zookeeper，也就是是否将信息注册到zookeeper上。默认为true。对于同步复制的slave来说，本参数会被强制设置为false。
• zk.zkConnect: zookeeper服务器列表，例如localhost:1281这样的字符串。默认也是localhost:2181。请设置你的zk集群地址列表。
• zk.zkSessionTimeoutMs: zookeeper的session timeout，默认为30秒。单位毫秒。
• zk.zkConnectionTimeoutMs： zookeeper的连接超时时间，默认同样为30秒，单位毫秒。
• zk.zkSyncTimeMs: 预期的zk集群间数据同步延迟，默认为5秒，这个参数对服务器无意义。

Topic配置

服务器将提供哪些topic服务都是通过topic配置来实现的，topic配置都是在[topic=xxx]的模块下面，其中xxx就是topic名称，一个示范配置如下：

[topic=boyan-test]

stat=true

numPartitions=1

这里配置了一个名为test的topic，并针对该topic启用实时统计，并将topic的在本服务器的分区数目设置为1。可见，topic配置可覆盖服务器的部分配置，包括：

• stat:是否启用实时统计，启用则会在服务端对该topic的请求做实时统计，可以通过stats topic-name协议观察到该topic运行状况，可选。
• numPartitions: 该topic在本服务器的分区总数，覆盖系统配置，可选。
• unflushInterval：每隔多少条消息做一次磁盘sync，覆盖系统配置，可选。
• unflushThreshold：每隔多少秒做一次磁盘sync，覆盖系统配置，可选。
• deletePolicy：topic的删除策略，覆盖系统配置，可选。
• deleteWhen：删除策略的执行时间，覆盖系统配置，可选。
• dataPath：设置数据文件路径，覆盖系统配置，可选。

1.4.3新增参数：

• acceptPublish: 是否接收该topic的消息，覆盖系统配置，可选。
• acceptSubscribe: 是否接受消费者的订阅，覆盖系统配置，可选。

新增Topic热部署

在新增或者删除topic并保存server.ini之后，可以通过下列命令热加载新的配置文件并生效：

bin/metaServer.sh reload

6.集群

集群

Meta假定producer、broker和consumer都是分布式的集群系统。

Producer可以是一个集群，多台机器上的producer可以往同一个topic发送消息。

Meta的服务器broker一般也是一个集群，多台broker组成一个集群提供一些topic服务，生产者按照一定的路由规则往集群里某台broker发送消息，消费者按照一定的路由规则拉取某台broker上的消息。

Consumer也可以组织成一个集群来消费同一个topic，发往这个topic的消息按照一定的路由规则发送到consumer集群里的某一台机器。Consumer集群每个consumer必须拥有相同的分组名称。

Broker集群配置

Broker集群配置非常容易，假设你已经按照如何开始和服务器配置管理配置好并启用了你第一台broker,某一天你发现这个单台broker无法支撑更大的消息量，那么你可能就需要引入更多的broker作为集群来提供服务，你要做的事情很简单：

• 拷贝broker1的配置文件conf/server.ini到新的broker,假设为broker2。
• 修改broker2的server.ini,只要修改brokerId为另一个不同于broker1的值即可
• 启动broker2，这样一来broker2将和broker1组成一个服务器集群
• 在这个过程中你不需要重启任何现有的服务，包括生产者、消费者和broker1，他们都将自动感知到新的broker2

可见，配置一个集群唯一要做的就是使用同一份配置文件并定义不同的brokerId即可。

负载均衡

负载均衡和failover分不开，我们将分别讨论下生产者和消费者的负载均衡策略。我们先假定broker是一个集群，这样每个topic必定有多个分区。

生产者的负载均衡和failover

每个broker都可以配置一个topic可以有多少个分区，但是在生产者看来，一个topic在所有broker上的的所有分区组成一个分区列表来使用。

在创建producer的时候，客户端会从zookeeper上获取publish的topic对应的broker和分区列表，生产者在发送消息的时候必须选择一台broker上的一个分区来发送消息，默认的策略是一个轮询的路由规则，一张图来表示

生产者在通过zk获取分区列表之后，会按照brokerId和partition的顺序排列组织成一个有序的分区列表，发送的时候按照从头到尾循环往复的方式选择一个分区来发送消息。考虑到我们的broker服务器软硬件配置基本一致，默认的轮询策略已然足够。

如果你想实现自己的负载均衡策略，可以实现上文提到过的PartitionSelector接口，并在创建producer的时候传入即可。

在broker因为重启或者故障等因素无法服务的时候，producer通过zookeeper会感知到这个变化，将失效的分区从列表中移除做到fail over。因为从故障到感知变化有一个延迟，可能在那一瞬间会有部分的消息发送失败。

消费者的负载均衡

消费者的负载均衡会相对复杂一些。我们这里讨论的是单个分组内的消费者集群的负载均衡，不同分组的负载均衡互不干扰，没有讨论的必要。消费者的负载均衡跟topic的分区数目紧密相关，要考察几个场景。 首先是，单个分组内的消费者数目如果比总的分区数目多的话，则多出来的消费者不参与消费，如图

其次，如果分组内的消费者数目比分区数目小，则有部分消费者要额外承担消息的消费任务，具体见示例图如下

综上所述，单个分组内的消费者集群的负载均衡策略如下

• 每个分区针对同一个group只挂载一个消费者
• 如果同一个group的消费者数目大于分区数目，则多出来的消费者将不参与消费
• 如果同一个group的消费者数目小于分区数目，则有部分消费者需要额外承担消费任务

Meta的客户端会自动帮处理消费者的负载均衡，它会将消费者列表和分区列表分别排序，然后按照上述规则做合理的挂载。

从上述内容来看，合理地设置分区数目至关重要。如果分区数目太小，则有部分消费者可能闲置，如果分区数目太大，则对服务器的性能有影响。

在某个消费者故障或者重启等情况下，其他消费者会感知到这一变化（通过 zookeeper watch消费者列表），然后重新进行负载均衡，保证所有的分区都有消费者进行消费。

7.HA

Ha介绍

Meta相比于kafka的一个重要特性就是消息高可用方案的实现，我们称之为HA方案。消息在发送到broker之后立即写入磁盘才返回客户端告诉消息生产者消息发送成功，通过unflushThreshold和unflushInterval两个参数的控制，可以保证单机消息数据的安全性，只要机器的磁盘没有永久损坏，消息总可以在重启后恢复并正常投递给消费者们。但是，如果遇到了磁盘永久损坏或者数据文件永久损坏的情况，那么该broker上的消息数据将可能永久丢失。为了防止这种情况的发生，一个可行的方案就是将消息数据复制到多台机器，类似mysql的主从复制功能。

同步复制和异步复制

meta提供类似mysql主从复制的异步复制和同步功能，分别对应不同的可靠级别。理论上说同步复制能带来更高的可靠级别，异步复制因为延迟的存在，可能会丢失极少量的消息数据，相应地，同步复制会带来性能的损失，因为要同步写入两台甚至更多的broker机器上才算写入成功。

在实际实践中，**我更推荐采用异步复制的架构**，因为异步复制的架构相对简单，并且易于维护和恢复，对性能也没有影响。而同步复制对运维要求相对很高，机制复杂容易出错，故障恢复也比较麻烦。**异步复制加上磁盘做磁盘阵列**，足以应对非常苛刻的数据可靠性要求。

异步复制配置

假设你已经根据如何开始这份文档配置了一台broker服务器，并且配置了一个topic为test，现在你希望test能复制到另一台slave broker上来保证消息数据的高可用。你可以这样做：

1.首先，你需要部署一个新的broker，具体仍然参照如何开始这份文档，配置server.ini从master broker拷贝一份。

2.其次，配置slave文件。编辑conf/async_slave.properties：

#slave编号,大于等于0表示作为slave启动,同一个master下的slave编号应该设不同值.

slaveId=0

#作为slave启动时向master订阅消息的group,如果没配置则默认为meta-slave-group

#不同的slaveId请使用不同的group

slaveGroup=meta-slave-group

#slave数据同步的最大延时,单位毫秒

slaveMaxDelayInMills=500

#是否自动从master同步server.ini, 1.4.2新增选项

#第一次仍然需要自己拷贝server.ini，后续可以通过设置此选项为true来自动同步

autoSyncMasterConfig=true

配置参数的含义请自己看注释。可见，一个master可以复制到多个slave。

3.执行下列命令启动slave:

bin/metaServer.sh start slave

4.第一次复制因为需要跟master完全同步需要耗费一定时间，你可以在数据文件的目录观察复制情况。

5.**请注意，异步复制的slave将参与消费者的消费活动，消息消费者可以从slave中获取消息并消费，消费者会随机从master和slaves中挑选一台作为消费broker。**

6.**请注意，从1.4.2开始，可以通过autoSyncMasterConfig选项配置是否自动同步master的server.ini到异步复制的slave上，当master的server.ini文件变更并通过bin/metaServer.sh reload之后，slave将监控到这一变更并自动同步。**

异步复制的局限

• 异步复制有延迟，虽然可以通过设定slaveMaxDelayInMills来控制延迟。

异步复制的故障处理

• Master永久故障： 将slave作为master启动，去除启动参数中的slave即可，也就是metaServer.sh restart
• Slave永久故障： 启动新的broker并配置作为master新的slave启动。

同步复制配置

TODO

第三部分 参考文档

网页：http://www.blogjava.net/killme2008/archive/2012/04/13/374110.html