MetaQ集群安装测试

package com.test.metaq;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.exception.MetaClientException;import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig;public class AsyncConsum {public static void main(String[] args) {final MetaClientConfig metaClientConfig = new MetaClientConfig();final ZKConfig zkConfig = new ZKConfig();zkConfig.zkConnect = "192.168.1.1:2181";metaClientConfig.setZkConfig(zkConfig);MessageSessionFactory sessionFactory = null;try {sessionFactory = new MetaMessageSessionFactory(metaClientConfig);} catch (MetaClientException e) {// TODO Auto-generated catch blocke.printStackTrace();}final String topic = "test";final String group = "meta-example";MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfig(group));try {consumer.subscribe(topic, 1024 * 1024, new MessageListener() {public void recieveMessages(Message message) {System.out.println("Receive message " + new String(message.getData()));}public Executor getExecutor() {return null;}});consumer.completeSubscribe();} catch (MetaClientException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}package com.test.metaq;import java.io.BufferedReader;import java.io.IOException;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.exception.MetaClientException;import com.taobao.metamorphosis.utils.ZkUtils.ZKConfig;public class Products {public static void main(String[] args) {final MetaClientConfig metaClientConfig = new MetaClientConfig();final ZKConfig zkConfig = new ZKConfig();zkConfig.zkConnect = "192.168.1.1:2181";metaClientConfig.setZkConfig(zkConfig);MessageSessionFactory sessionFactory = null;try {sessionFactory = new MetaMessageSessionFactory(metaClientConfig);} catch (MetaClientException e) {e.printStackTrace();}MessageProducer producer = sessionFactory.createProducer();final String topic = "test";producer.publish(topic);BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));String line = "qiujinyong";try {while ((line = reader.readLine()) != null) {SendResult sendResult = producer.sendMessage(new Message(topic,line.getBytes()));if (!sendResult.isSuccess()) {System.err.println("Send message failed,error message:"+ sendResult.getErrorMessage());} else {System.out.println("Send message successfully,sent to "+ sendResult.getPartition());}}} catch (IOException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (MetaClientException e) {// TODO Auto-generated catch blocke.printStackTrace();} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}}}　　打包test.jar后，传服务器上 java -cp test.jar com.test.metaq.Products 命令行输入message　　打包test.jar后，传服务器上 java -cp test.jar com.test.metaq.AsyncConsum 命令行会接收到message

5，应用例子