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客户端代码 生产者: Java代码
1. package com.metaq.product; 2.
3. import java.io.BufferedReader; 4. import java.io.InputStreamReader; 5.
6. import com.taobao.metamorphosis.Message;
7. import com.taobao.metamorphosis.client.MessageSessionFactory; 8. import com.taobao.metamorphosis.client.MetaClientConfig;
9. import com.taobao.metamorphosis.client.MetaMessageSessionFactory; 10. import com.taobao.metamorphosis.client.producer.MessageProducer; 11. import com.taobao.metamorphosis.client.producer.SendResult; 12. import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig; 13. public class Products {
14. public static void main(String[] args) throws Exception {
15. final MetaClientConfig metaClientConfig = new MetaClientConfig(); 16. final ZKConfig zkConfig = new ZKConfig(); 17. zkConfig.zkConnect = \18. metaClientConfig.setZkConfig(zkConfig);
19. // 由这个工厂创建生产者或者消费者
20. //1.服务的查找和发现,通过diamond和zookeeper帮你查找日常的meta服
务器地址列表
21. //2.连接的创建和销毁,自动创建和销毁到meta服务器的连接,并做连接复
用,也就是到同一台meta的服务器在一个工厂内只维持一个连接。 22. //3.消息消费者的消息存储和恢复,后续我们会谈到这一点。 23. //4.协调和管理各种资源,包括创建的生产者和消费者的。
24. MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaC
lientConfig);
25. //消息生产者的接口,MessageProducer是线程安全的,MessageProducer创建
的代价昂贵,每次都需要通过zk
26. //查找服务器并创建tcp长连接,通过它来发送消息,每个消息对象都是
Message类的实例,Message表示一个消息对象,它包含这么几个属性:
27. //id: Long型的消息id,消息的唯一id,系统自动产生,用户无法设置,在发送
成功后由服务器返回,发送失败则为0。
28. //topic: 消息的主题,订阅者订阅该主题即可接收发送到该主题下的消息,生
产者通过指定发布的topic查找到需要连接的服务器地址,必须。
29. //data: 消息的有效载荷,二进制数据,也就是消息内容,meta永远不会修改
消息内容,你发送出去是什么样子,接收到就是什么样子。 30. //消息内容通常限制在1M以内,我的建议是最好不要发送超过上百K的消息,
必须。数据是否压缩也完全取决于用户。
31. //attribute: 消息属性,一个字符串,可选。发送者可设置消息属性来让消费者
过滤。
32. MessageProducer producer = sessionFactory.createProducer(); 33. final String topic = \34. producer.publish(topic);
35. BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)
);
36. String line = \
37. while ((line = reader.readLine()) != null) { 38. // send message
39. SendResult sendResult = producer.sendMessage(new Message(topic, line.getByt
es()));
40. // check result
41. if (!sendResult.isSuccess()) {
42. System.err.println(\
rMessage()); 43. } 44. else {
45. System.out.println(\
tion()); 46. } 47. } 48. } 49. }
消费者: Java代码
1. package com.metaq.consum; 2.
3. import java.util.concurrent.Executor; 4.
5. import com.taobao.metamorphosis.Message;
6. import com.taobao.metamorphosis.client.MessageSessionFactory; 7. import com.taobao.metamorphosis.client.MetaClientConfig;
8. import com.taobao.metamorphosis.client.MetaMessageSessionFactory; 9. import com.taobao.metamorphosis.client.consumer.ConsumerConfig; 10. import com.taobao.metamorphosis.client.consumer.MessageConsumer; 11. import com.taobao.metamorphosis.client.consumer.MessageListener; 12. import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig; 13.
14. public class AsyncConsum {
15. public static void main(String[] args) throws Exception {
16. final MetaClientConfig metaClientConfig = new MetaClientConfig();
17. final ZKConfig zkConfig = new ZKConfig(); 18. zkConfig.zkConnect = \19. metaClientConfig.setZkConfig(zkConfig);
20. MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClie
ntConfig);
21. final String topic = \
22. final String group = \
23. //每个消息者都必须有一个ConsumerConfig配置对象,我们这里设置了group属
性,这是消费者的分组名称
24. //Meta的Producer、Consumer和Broker都可以为集群。消费者可以组成一个集
群共同消费同一个topic,
25. //发往这个topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。
同一个消费者集群必须拥有同
26. //一个分组名称,也就是同一个group。我们这里将分组名称设置为
meta-example
27. MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfi
g(group)); 28. //topic,订阅的主题
29. //maxSize,因为meta是一个消费者主动拉取的模型,这个参数规定每次拉取的
最大数据量,单位为字节,这里设置为1M,默认最大为1M。 30. //MessageListener,消息监听器,负责消息消息。
31. consumer.subscribe(topic, 1024 * 1024, new MessageListener() { 32. public void recieveMessages(Message message) {
33. System.out.println(\34. }
35. //消息的消费过程可以是一个并发处理的过程,getExecutor返回你想设置的线
程池,每次消费都会
36. //在这个线程池里进行。recieveMessage方法用于实际的消息消费处理,
message参数即为消费者收到的消息,它必不为null。 37. //我们这里简单地打印收到的消息内容就完成消费。如果在消费过程中抛出任
何异常,该条消息将会
38. //在一定间隔后重新尝试提交给MessageListener消费。在多次消费失败的情
况下,该消息将会存储到消费者应用的本次磁盘, 39. //并在后台自动恢复重试消费 40. public Executor getExecutor() { 41. return null; 42. } 43. });
44. // 在调用subscribe之后,我们还调用了completeSubscribe方法来完成订阅过
程。请注意,
45. //subscribe仅是将订阅信息保存在本地,并没有实际跟meta服务器交互,要
使得订阅关系生效必须调用
46. //一次completeSubscribe,completeSubscribe仅能被调用一次,多次调用将抛
出异常。 为什么需
47. //要completeSubscribe方法呢,原因有二: 48. //首先,subscribe方法可以被调用多次,也就是一个消费者可以消费多种topic 49. //其次,如果每次调用subscribe都跟zk和meta服务器交互一次,代价太高 50. //因此completeSubscribe一次性将所有订阅的topic生效,并处理跟zk和meta
服务器交互的所有过程。
51. // 同样,MessageConsumer也是线程安全的,创建的代价不低,因此也应该尽
量复用
52. consumer.completeSubscribe(); 53. } 54. 55. }
启动: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一开始就是为大规模分布式系统设计的,如果不当使用,可能没有带来好处,反而多出一堆问题。开发者需要根据自己面对的场景,团队的技术能力,做出一个合适的选择。