0
点赞
收藏
分享

微信扫一扫

metamorphosis-3-metaQ初步


Java客户端例子

使用maven,引用metaq的java client非常简单:


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


 

 也可以引用 client-extend


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

 

直接用git 将metamorphosis-example clone  下载页面:https://github.com/killme2008/metamorphosis-example 。

请注意,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服务器地址列表,diamond可以忽略。

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

3.消费者的消息存储和恢复。

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

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

 


MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(new

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

 

消息生产者  

 

 

package
  

import
  
import
  

import
  
import
  
import
  
import
  
import
  
import
  
import
  

public class
  
public static void main(String[] args) throws
  
final MetaClientConfig metaClientConfig = new
  
final ZKConfig zkConfig = new
  
//设置zookeeper地址

  
        metaClientConfig.setZkConfig(zkConfig);
  
// New session factory,强烈建议使用单例
        MessageSessionFactory sessionFactory = new
  
// create producer,强烈建议使用单例

  
// publish topic
        final
  
        producer.publish(topic);
  

new BufferedReader(new
  
null;
  
while ((line = reader.readLine()) != null) {
  
// send message
            SendResult sendResult = producer.sendMessage(new
  
// check result
            if
  
                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(另外,metaq的broker必须要先通过配置文件设置好topic,启动broker后,客户端才可能发布消息成功,不然就会报出异常),这是为了让会话工厂帮你去查找接收这些topic的meta服务器地址并初始化连接。这个步骤针对每个topic只需要做一次,多次调用无影响。

总结下这个例子,从标准输入读入你输入的数据,并将数据封装成一个Message对象,发送到[localhost:8123]服务的 topic为metatestbroker 上。

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

消息消费者

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

package
  

import
  

import
  
import
  
import
  
import
  
import
  
import
  
import
  
import
  


public class
  
public static void main(String[] args) throws
  
final MetaClientConfig metaClientConfig = new
  
final ZKConfig zkConfig = new
  
//设置zookeeper地址

  
        metaClientConfig.setZkConfig(zkConfig);
  
// New session factory,强烈建议使用单例
        MessageSessionFactory sessionFactory = new
  
// subscribed topic
        final
  
// consumer group
        final
  
// create consumer,强烈建议使用单例
        MessageConsumer consumer = sessionFactory.createConsumer(new
  
// subscribe topic
        consumer.subscribe(topic, 1024 * 1024, new
  

public void
  
new
  
            }
  


public
  
// Thread pool to process messages,maybe null.
                return null;
  
            }
  
        });
  
// complete subscribe

  
    }
  
}



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

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

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

MessageListener的接口方法如下:

 


public interface
  
/**
     * 接收到消息列表,只有message不为空并且不为null的情况下会触发此方法
     * 
     * @param message
     */
  
public void
  


/**
     * 处理消息的线程池
     * 
     * @return
     */
  
public
  
}


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

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

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

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

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

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

举报

相关推荐

0 条评论