一、maven依赖

<dependency><groupId>com.taobao.metamorphosis</groupId><artifactId>metamorphosis-client</artifactId><version>1.4.6.2</version>
</dependency><dependency><groupId>com.taobao.metamorphosis</groupId><artifactId>metamorphosis-client-extension</artifactId><version>1.4.6.2</version>
</dependency>

二、发送者

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;import java.io.BufferedReader;
import java.io.InputStreamReader;/*** Created by lc-t123 on 2016/4/14.*/
public class Producer {public static void main(String[] args) throws Exception {final MetaClientConfig metaClientConfig = new MetaClientConfig();final ZKConfig zkConfig = new ZKConfig();//设置zookeeper地址zkConfig.zkConnect = "192.168.1.70:2181";metaClientConfig.setZkConfig(zkConfig);// New session factory,强烈建议使用单例MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);/**  create producer,强烈建议使用单例*  消息生产者的接口是MessageProducer，你可以通过它来发送消息*/MessageProducer producer = sessionFactory.createProducer();// publish topicfinal String topic = "test";/** 这一步在发送消息前是必须的，你必须发布你将要发送消息的topic* 这是为了让会话工厂帮你去查找接收这些topic的meta服务器地址并初始化连接* 这个步骤针对每个topic只需要做一次，多次调用无影响*/producer.publish(topic);BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));String line = null;while ((line = reader.readLine()) != null){/** send message* 在Meta里，每个消息对象都是Message类的实例，Message表示一个消息对象，它包含这么几个属性：* 1) id: Long型的消息id,消息的唯一id，系统自动产生，用户无法设置，在发送成功后由服务器返回，发送失败则为0。* 2) topic: 消息的主题，订阅者订阅该主题即可接收发送到该主题下的消息，生产者通过指定发布的topic查找到需要连接的服务器地址，必须。* 3) data: 消息的有效载荷，二进制数据，也就是消息内容，meta永远不会修改消息内容，你发送出去是什么样子，接收到就是什么样子。消息内容通常限制在1M以内，我的建议是最好不要发送超过上百K的消息，必须。数据是否压缩也完全取决于用户。* 4) attribute: 消息属性，一个字符串，可选。发送者可设置消息属性来让消费者过滤。*/SendResult sendResult = producer.sendMessage(new Message(topic, line.getBytes()));// check resultif (!sendResult.isSuccess()){System.err.println("Send message failed,error message:" + sendResult.getErrorMessage());}else {System.out.println("Send message successfully,sent to " + sendResult.getPartition());}}}
}

三、接收者

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;import java.util.concurrent.Executor;public class AsyncConsumer {public static void main(String[] args) throws Exception {final MetaClientConfig metaClientConfig = new MetaClientConfig();final ZKConfig zkConfig = new ZKConfig();//设置zookeeper地址zkConfig.zkConnect = "192.168.1.70:2181";metaClientConfig.setZkConfig(zkConfig);// New session factory,强烈建议使用单例MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);// subscribed topicfinal String topic = "test";// consumer groupfinal String group = "meta-example";/** create consumer,强烈建议使用单例* 通过createConsumer方法来创建MessageConsumer，注意到我们传入一个ConsumerConfig参数，* 这是消费者的配置对象。每个消息者都必须有一个ConsumerConfig配置对象，* 我们这里只设置了group属性，这是消费者的分组名称。* Meta的Producer、Consumer和Broker都可以为集群。* 消费者可以组成一个集群共同消费同一个topic，发往这个topic的消息将按照一定的负载均衡规则发送给集群里的一台机器。* 同一个消费者集群必须拥有同一个分组名称，也就是同一个group。我们这里将分组名称设置为meta-example*/MessageConsumer consumer = sessionFactory.createConsumer(new ConsumerConfig(group));/** subscribe topic* 订阅消息通过subscribe方法，这个方法接受三个参数* 1) topic，订阅的主题* 2) maxSize，因为meta是一个消费者主动拉取的模型，这个参数规定每次拉取的最大数据量，单位为字节，这里设置为1M，默认最大为1M。* 3) MessageListener，消息监听器，负责消息消息。*/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 subscribeconsumer.completeSubscribe();}
}

metaq-server安装参考官方文档

可以通过http://192.168.1.70:8120/ 访问web界面