MetaQ初探

 

 

MetaQ（全称Metamorphosis）是一个高性能、高可用、可扩展的分布式消息中间件，,MetaQ具有消息存储顺序写、吞吐量大和支持本地和XA事务等特性，适用于大吞吐量、顺序消息、广播和日志数据传输等场景，METAQ在阿里巴巴各个子公司被广泛应用，每天转发250亿+条消息。主要应用于异步解耦，Mysql数据复制，收集日志等场景。

总体结构

 

主要特点

• 生产者、服务器和消费者都可分布式
• 消息存储顺序写
• 性能极高,吞吐量大
• 支持消息顺序
• 支持本地和XA事务
• 客户端pull，随机读,利用sendfile系统调用，zero-copy ,批量拉数据
• 支持消费端事务
• 支持消息广播模式
• 支持异步发送消息
• 支持http协议
• 支持消息重试和recover
• 数据迁移、扩容对用户透明
• 消费状态保存在客户端
• 支持同步和异步复制两种HA

主要特性

数据完整性

消息生产者发送的消息，meta服务器收到后在做必要的校验和检查之后的第一件事就是写入磁盘，写入成功之后返回应答给生产者，生产者发送消息返回SendResult，如果isSuccess返回为true,则表示消息已经确认发送到服务器并被服务器接收存储。整个发送过程是一个同步的过程。保证消息送达服务器并返回结果。因此，可以确认每条发送结果为成功的消息服务器都是写入磁盘的。

写入磁盘，不意味着数据落到磁盘设备上，毕竟我们还隔着一层os，os对写有缓冲。Meta有两个特性来保证数据落到磁盘上:每1000条（可配置），即强制调用一次force来写入磁盘设备。每隔10秒（可配置），强制调用一次force来写入磁盘设备。因此,Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。当然通过参数调整你甚至可以在掉电情况下不丢失任何消息。

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

数据可靠性

服务器通常组织为一个集群，一条从生产者过来的消息可能按照路由规则存储到集群中的某台机器。Meta已经实现高可用的HA方案，类似mysql的同步和异步复制，将一台meta服务器的数据完整复制到另一台slave服务器，并且slave服务器还提供消费功能（同步复制不提供消费）。消息的消费者是一条接着一条地消费消息，只有在成功消费一条消息后才会接着消费下一条。如果在消费某条消息失败（如异常），则会尝试重试消费这条消息（默认最大5次），超过最大次数后仍然无法消费，则将消息存储在消费者的本地磁盘，由后台线程继续做重试。而主线程继续往后走，消费后续的消息。因此，只有在MessageListener确认成功消费一条消息后，meta的消费者才会继续消费另一条消息。由此来保证消息的可靠消费。

消费者的另一个可靠性的关键点是offset的存储，也就是拉取数据的偏移量。我们目前提供了以下几种存储方案zookeeper，默认存储在zoopkeeper上，zookeeper通过集群来保证数据的安全性。mysql，可以连接到您使用的mysql数据库，只要建立一张特定的表来存储。完全由数据库来保证数据的可靠性。file，文件存储，将offset信息存储在消费者的本地文件中。Offset会定期保存，并且在每次重新负载均衡前都会强制保存一次

 

下载、配置、运行

 

https://code.google.com/p/meta-queue/downloads/list选择最新版本的服务器并下载到本地解压缩文件,bin目录存放的脚本文件，日志在logs目录，而配置文件主要是conf目录下server.ini，lib存放所有的依赖jar包。

进入bin/env.sh，修改JAVA_HOME,JMX等变量。

根据需要修改conf/server.ini文件(列出了所有的配置)：

zk.zkEnable=true是否注册到zk，默认为true
zk.zkConnect=localhost:2181  zk的服务器列表
zk.zkSessionTimeoutMs=30000 zk心跳超时，单位毫秒，默认30秒 zk.zkSessionTimeoutMs=30000
zk.zkConnectionTimeoutMs=30000  zk连接超时时间，单位毫秒，默认30秒
zk.zkSyncTimeMs=5000 zk数据同步时间，单位毫秒，默认5秒
brokerId：服务器ID(必须是集群内唯一) 
serverPort：服务器端口
hostName：默认将取本机IP (多机网卡，需要指明)
dataLogPath：日志数据文件路径，默认跟dataPath一样
dataPath:于指定默认的数据存储路径（慎重设置，默认在user.home/meta下）
numPartitions:默认topic的分区数目（慎重设置）
maxSegmentSize：单个文件的最大大小,实际会超过此值，默认1G
maxTransferSize：传输给客户端每次最大的缓冲区大小，默认1M
unflushThreshold：最大允许的未flush间隔时间，毫秒，默认10秒
putProcessThreadCount：;处理put请求线程数，默认cpus*10
deletePolicy=delete,168(数据删除策略，默认超过7天即删除,这里的168是小时，10s表示10秒，10m表示10分钟，10h表示10小时，默认为小时)
deleteWhen: 何时执行删除策略的cron表达式，默认是0 0 6,18 * * ?，也就是每天的早晚6点执行处理策略。deleteWhen: 删除策略的执行时间，cron表达式
maxCheckpoints: 最大保存事务checkpoint数目，默认为3
checkpointInterval: 事务checkpoint时间间隔，单位毫秒，默认1小时(3600000)
maxTxTimeoutTimerCapacity=30000最大事务超时事件数，用于监控事务超时
maxTxTimeoutInSeconds=60最大事务超时时间，单位秒
flushTxLogAtCommit=1事务日志的同步设置，0表示让操作系统决定，1表示每次commit都同步，2表示每隔1秒同步一次，此参数严重影响事务性能，可根据你需要的性能和可靠性之间权衡做出一个合理的选择。通常建议设置为2，表示每隔1秒刷盘一次，也就是最多丢失一秒内的运行时事务。这样的可靠级别对大多数服务是足够的。最安全的当然是设置为1，但是将严重影响事务性能。而0的安全级别最低。安全级别上 1>=2>0，而性能则是0 >= 2 > 1。

diamondZKDataId=metamorphosis.zkConfig   zk在diamond中配置存储的dataId
diamondZKGroup=DEFAULT_GROUP  zk在diamond中配置存储的group
acceptPublish: 是否接收消息，默认为true；如果为false，则不会注册发送信息到zookeeper上，客户端当然无法发送消息到该broker。本参数可以被后续的topic配置覆盖。
acceptSubscribe: 与acceptPublish类似，默认也为true；如果为false，则不会注册消费信息到zookeeper上，消费者无法发现该broker，当然无法从该broker消费消息。本参数可以被后续的topic配置覆盖。

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

JAVA客户端代码

生产者：

package com.metaq.product;

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 Products {
        public static void main(String[] args) throws Exception {
            final MetaClientConfig metaClientConfig = new MetaClientConfig();
            final ZKConfig zkConfig = new ZKConfig();
            zkConfig.zkConnect = "192.168.2.11:2181";
            metaClientConfig.setZkConfig(zkConfig);
            // 由这个工厂创建生产者或者消费者
            //1.服务的查找和发现，通过diamond和zookeeper帮你查找日常的meta服务器地址列表
            //2.连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台meta的服务器在一个工厂内只维持一个连接。
            //3.消息消费者的消息存储和恢复，后续我们会谈到这一点。
            //4.协调和管理各种资源，包括创建的生产者和消费者的。
            MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
            //消息生产者的接口，MessageProducer是线程安全的,MessageProducer创建的代价昂贵，每次都需要通过zk
            //查找服务器并创建tcp长连接,通过它来发送消息，每个消息对象都是Message类的实例，Message表示一个消息对象，它包含这么几个属性：
            //id: Long型的消息id,消息的唯一id，系统自动产生，用户无法设置，在发送成功后由服务器返回，发送失败则为0。
            //topic: 消息的主题，订阅者订阅该主题即可接收发送到该主题下的消息，生产者通过指定发布的topic查找到需要连接的服务器地址，必须。
            //data: 消息的有效载荷，二进制数据，也就是消息内容，meta永远不会修改消息内容，你发送出去是什么样子，接收到就是什么样子。
            //消息内容通常限制在1M以内，我的建议是最好不要发送超过上百K的消息，必须。数据是否压缩也完全取决于用户。
            //attribute: 消息属性，一个字符串，可选。发送者可设置消息属性来让消费者过滤。
            MessageProducer producer = sessionFactory.createProducer();
            final String topic = "test";
            producer.publish(topic);
            BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
            String line = "qiujinyong";
            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());
                }
            }
        }
    }

package com.metaq.product;

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 Products {
	    public static void main(String[] args) throws Exception {
	        final MetaClientConfig metaClientConfig = new MetaClientConfig();
	        final ZKConfig zkConfig = new ZKConfig();
	        zkConfig.zkConnect = "192.168.2.11:2181";
	        metaClientConfig.setZkConfig(zkConfig);
	        // 由这个工厂创建生产者或者消费者
	        //1.服务的查找和发现，通过diamond和zookeeper帮你查找日常的meta服务器地址列表
	        //2.连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台meta的服务器在一个工厂内只维持一个连接。
	        //3.消息消费者的消息存储和恢复，后续我们会谈到这一点。
	        //4.协调和管理各种资源，包括创建的生产者和消费者的。
	        MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
	        //消息生产者的接口，MessageProducer是线程安全的,MessageProducer创建的代价昂贵，每次都需要通过zk
	        //查找服务器并创建tcp长连接,通过它来发送消息，每个消息对象都是Message类的实例，Message表示一个消息对象，它包含这么几个属性：
	        //id: Long型的消息id,消息的唯一id，系统自动产生，用户无法设置，在发送成功后由服务器返回，发送失败则为0。
	        //topic: 消息的主题，订阅者订阅该主题即可接收发送到该主题下的消息，生产者通过指定发布的topic查找到需要连接的服务器地址，必须。
	        //data: 消息的有效载荷，二进制数据，也就是消息内容，meta永远不会修改消息内容，你发送出去是什么样子，接收到就是什么样子。
	        //消息内容通常限制在1M以内，我的建议是最好不要发送超过上百K的消息，必须。数据是否压缩也完全取决于用户。
	        //attribute: 消息属性，一个字符串，可选。发送者可设置消息属性来让消费者过滤。
	        MessageProducer producer = sessionFactory.createProducer();
	        final String topic = "test";
	        producer.publish(topic);
	        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
	        String line = "qiujinyong";
	        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());
	            }
	        }
	    }
	}

 

消费者：

package com.metaq.consum;

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 AsyncConsum {
    public static void main(String[] args) throws Exception {
        final MetaClientConfig metaClientConfig = new MetaClientConfig();
        final ZKConfig zkConfig = new ZKConfig();
        zkConfig.zkConnect = "192.168.2.11:2181";
        metaClientConfig.setZkConfig(zkConfig);
        MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
        final String topic = "test";
        final String group = "meta-example";
        //每个消息者都必须有一个ConsumerConfig配置对象,我们这里设置了group属性，这是消费者的分组名称
        //Meta的Producer、Consumer和Broker都可以为集群。消费者可以组成一个集群共同消费同一个topic，
        //发往这个topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。同一个消费者集群必须拥有同
        //一个分组名称，也就是同一个group。我们这里将分组名称设置为meta-example
        MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfig(group));
        //topic，订阅的主题
        //maxSize，因为meta是一个消费者主动拉取的模型，这个参数规定每次拉取的最大数据量，单位为字节，这里设置为1M，默认最大为1M。
        //MessageListener，消息监听器，负责消息消息。
        consumer.subscribe(topic, 1024 * 1024, new MessageListener() {
            public void recieveMessages(Message message) {
                System.out.println("Receive message " + new String(message.getData()));
            }
            //消息的消费过程可以是一个并发处理的过程，getExecutor返回你想设置的线程池，每次消费都会
            //在这个线程池里进行。recieveMessage方法用于实际的消息消费处理，message参数即为消费者收到的消息，它必不为null。
            //我们这里简单地打印收到的消息内容就完成消费。如果在消费过程中抛出任何异常，该条消息将会
            //在一定间隔后重新尝试提交给MessageListener消费。在多次消费失败的情况下，该消息将会存储到消费者应用的本次磁盘，
            //并在后台自动恢复重试消费
            public Executor getExecutor() {
                return null;
            }
        });
            // 在调用subscribe之后，我们还调用了completeSubscribe方法来完成订阅过程。请注意，
            //subscribe仅是将订阅信息保存在本地，并没有实际跟meta服务器交互，要使得订阅关系生效必须调用
            //一次completeSubscribe，completeSubscribe仅能被调用一次，多次调用将抛出异常。 为什么需
            //要completeSubscribe方法呢，原因有二：
            //首先，subscribe方法可以被调用多次，也就是一个消费者可以消费多种topic
            //其次，如果每次调用subscribe都跟zk和meta服务器交互一次，代价太高
            //因此completeSubscribe一次性将所有订阅的topic生效，并处理跟zk和meta服务器交互的所有过程。
            // 同样,MessageConsumer也是线程安全的，创建的代价不低，因此也应该尽量复用
             consumer.completeSubscribe();
    }

}

package com.metaq.consum;

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 AsyncConsum {
    public static void main(String[] args) throws Exception {
        final MetaClientConfig metaClientConfig = new MetaClientConfig();
        final ZKConfig zkConfig = new ZKConfig();
        zkConfig.zkConnect = "192.168.2.11:2181";
        metaClientConfig.setZkConfig(zkConfig);
        MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
        final String topic = "test";
        final String group = "meta-example";
        //每个消息者都必须有一个ConsumerConfig配置对象,我们这里设置了group属性，这是消费者的分组名称
        //Meta的Producer、Consumer和Broker都可以为集群。消费者可以组成一个集群共同消费同一个topic，
        //发往这个topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。同一个消费者集群必须拥有同
        //一个分组名称，也就是同一个group。我们这里将分组名称设置为meta-example
        MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfig(group));
        //topic，订阅的主题
        //maxSize，因为meta是一个消费者主动拉取的模型，这个参数规定每次拉取的最大数据量，单位为字节，这里设置为1M，默认最大为1M。
        //MessageListener，消息监听器，负责消息消息。
        consumer.subscribe(topic, 1024 * 1024, new MessageListener() {
            public void recieveMessages(Message message) {
                System.out.println("Receive message " + new String(message.getData()));
            }
            //消息的消费过程可以是一个并发处理的过程，getExecutor返回你想设置的线程池，每次消费都会
            //在这个线程池里进行。recieveMessage方法用于实际的消息消费处理，message参数即为消费者收到的消息，它必不为null。
            //我们这里简单地打印收到的消息内容就完成消费。如果在消费过程中抛出任何异常，该条消息将会
            //在一定间隔后重新尝试提交给MessageListener消费。在多次消费失败的情况下，该消息将会存储到消费者应用的本次磁盘，
            //并在后台自动恢复重试消费
            public Executor getExecutor() {
                return null;
            }
        });
            // 在调用subscribe之后，我们还调用了completeSubscribe方法来完成订阅过程。请注意，
            //subscribe仅是将订阅信息保存在本地，并没有实际跟meta服务器交互，要使得订阅关系生效必须调用
            //一次completeSubscribe，completeSubscribe仅能被调用一次，多次调用将抛出异常。 为什么需
            //要completeSubscribe方法呢，原因有二：
            //首先，subscribe方法可以被调用多次，也就是一个消费者可以消费多种topic
            //其次，如果每次调用subscribe都跟zk和meta服务器交互一次，代价太高
            //因此completeSubscribe一次性将所有订阅的topic生效，并处理跟zk和meta服务器交互的所有过程。
            // 同样,MessageConsumer也是线程安全的，创建的代价不低，因此也应该尽量复用
             consumer.completeSubscribe();
    }

}

 

启动：bin/metaServer.sh start  （更多的命令直接help一下。）观察：日志方式（logs/metaServer.log），命令方式（bin/metaServer.sh stats）,telnet方式（telnet IP端口 stats）

启动服务器后执行客户端的生产者和消费者的代码，运行结果如下（生产者随便在控制台输入消息后，进行回车，生产者消息如果成功发送到broker，会返回一条发送成功的消息，同时消费者会接收到该消息）：

生产者

消费者

 

集群

启动metaQ后，它将启动一个内置的zookeeper，并将broker注册到该zookeeper。但MetaQ应该是作为一个分布式集群提供服务。MetaQ的集群管理是利用zookeeper实现的，使用zookeeper发布和订阅服务，并默认使用zookeeper存储消费者offset，你需要首先安装一个zookeeper到某台机器上，或者使用某个现有的zk集群，然后配置zookeeper（zk配置参见我blog《zookeeper初探》）

负载均衡

每个broker都可以配置一个topic可以有多少个分区，但是在生产者看来，一个topic在所有broker上的的所有分区组成一个分区列表来使用。在创建producer的时候，客户端会从zookeeper上获取publish的topic对应的broker和分区列表，生产者在发送消息的时候必须选择一台broker上的一个分区来发送消息，默认的策略是一个轮询的路由规则如果你想实现自己的负载均衡策略，可以实现上文提到过的PartitionSelector接口，并在创建producer的时候传入即可对于消费者而言，合理地设置分区数目至关重要。如果分区数目太小，则有部分消费者可能闲置，如果分区数目太大，则对服务器的性能有影响。在某个消费者故障或者重启等情况下，其他消费者会感知到这一变化（通过 zookeeper watch消费者列表），然后重新进行负载均衡，保证所有的分区都有消费者进行消费。拷贝broker1的配置文件conf/server.ini到新的broker,假设为broker2。修改broker2的server.ini,只要修改brokerId为另一个不同于broker1的值即可，启动broker2， 在这个过程中你不需要重启任何现有的服务，包括生产者、消费者和broker1，他们都将自动感知到新的broker2

主从复制

先配置负载均衡后（和上面配置一样），然后再配置从机的另一个文件(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

 

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

评论

1 楼 xiaguobing 2013-08-19  

谢谢分享啊