在上篇文章，阿堂和大家分享了《分布式消息队列中间件系列研究之阿堂教程(基础篇-Local模式)》，后面由于时间关系，就一直没有接着写了。忙里偷闲，昨天晚上在家写了大部分，今天抽点时间阿堂将继续为大家奉献完成《分布式消息队列中间件系列研究之阿堂教程(进阶篇)》。这里阿堂结合发送邮件的一个相对具体的案例，应该说是有一定的代表性的，相对比较深入的剖析开源框架metq分布式消息队列的使用。相信通过阿堂的分享后，大家就基本明白分布式消息队列是怎么回事和大致知道如何使用了。当然，网友们如果想非常深入的学习和使用metq，建议网友们可以直接到metq的官网去学习和了解。

metq使用的大致流程如下所示

public class Productor {

private static Log log = LogFactory.getLog(Productor.class);

public static void main(String[] args) throws Exception {

//由消息工厂产生消息生产者

MessageProducer producer = MessageSessionFactoryManager.getSessionFactory(true).createProducer();
//设置topic,必须要在server.ini文件中进行配置
final String topic = "email";
//发布topic
producer.publish(topic);
//发布内容(根据实际业务来组装内容)
String line = "网络时空(阿堂)恭喜大家2015年心想事成!";
try{
//模拟发邮件(根据实际业务定义，可以是发邮件，发短信，httpPost提交等均可)
EmailRequest request = new EmailRequest("", topic,line);
//序列化发送内容
String jsonString = JSON.toJSONString(request, SerializerFeature.WriteClassName);
log.info("------------------------------------------------------------------------");
log.info("发布者发送的EmailAddress = "+request.getEmailAddress());
log.info("发布者发送的 Topic = "+request.getEmailTopic());
log.info("发布者发送的邮件的内容= "+request.getContent());
log.info("发布者发送的邮件发送的时间点= "+request.getCreateTime());
log.info("------------------------------------------------------------------------");
//发布订消息，这里老何定义了重写回调SendMessageCallbackImpl实现方法
producer.sendMessage(new Message(topic, jsonString.getBytes()),new SendMessageCallbackImpl(producer,topic,jsonString));
}catch(Exception ex){
log.error(ex);
}

}

}

------------------------------------

public class SendMessageCallbackImpl implements SendMessageCallback {

private static Log log = LogFactory.getLog(SendMessageCallbackImpl.class);

private MessageProducer messageProducer;
private String topic;
private String content;
public SendMessageCallbackImpl( MessageProducer messageProducer, String topic, String content) {
super();
this.messageProducer = messageProducer;
this.topic = topic;
this.content = content;
}

public void onException(Throwable e) {
log.fatal("metaq server exception , error message:" + e.getMessage());
log.info("--------------------------------------------------------------");
//出现异常时写入日志文件，进行补偿机制
logToFile();
log.info("--------------------------------------------------------------");
}

public void onMessageSent(SendResult result) {
log.info("result = "+result);
if (!result.isSuccess()) {
log.info("--------------------------------------------------------------");
log.warn("Send " + topic + " message failed,error message:" + result.getErrorMessage());
//没有收到broker服务器的正常应答时写入日志文件，进行补偿机制
logToFile();
}
else {
log.info("--------------------------------------------------------------");
log.info("Send "+topic+" successfully,sent to " + result.getPartition()+" "+result.getOffset());
log.info("--------------------------------------------------------------");
}
}
private void logToFile() {
//定义日志文件
String fileName = MessageSessionFactoryManager.getMessagedir()+UUID.randomUUID().toString();
File file = new File(fileName);
FileWriter fw = null;
try {
fw = new FileWriter(file);
fw.write(topic+"\r\n");
fw.write(content);
fw.flush();
} catch (IOException e) {
log.error(e);
} finally {
if(fw != null) {
try {
fw.close();
} catch (IOException e) {
}
}
}
}

}

------------------------------------

public class Consumer {

private static Log log = LogFactory.getLog(Consumer.class);
public static void main(String[] args) throws Exception {

final String topic = "email";

final String group = "meta-example";
try{
//复用SessionFacotory(单例模式) 发布者和订阅者共用sessionFactory
MessageConsumer consumer = MessageSessionFactoryManager.getSessionFactory(false).createConsumer(new ConsumerConfig(group));
//每次订阅300k的字节流内容，将订阅信息保存到本地
consumer.subscribe(topic, 1024 * 300, new EmailMessageListener());
//completeSubscribe一次性将所有的订阅生效，并处理zk和metaq服务器的所有交互过程
consumer.completeSubscribe();
}catch(Exception ex){
log.error(ex);
}

}

}

------------------------------------

public class EmailMessageListener implements MessageListener {
private static Log log = LogFactory.getLog(EmailMessageListener.class);

public void recieveMessages(Message message) throws InterruptedException {

try {
log.info("------------------------------------------------------------------------");
log.info("Receive Email message, BrokerId-Partition:"
+ message.getPartition().getBrokerId() + ","
+ message.getPartition().getPartition()+", "+message.getTopic()+" ,"+message.getId());
//反序列化接收内容
EmailRequest emailRequest = JSON.parseObject(new String(message.getData()), EmailRequest.class);
log.info("------------------------------------------------------------------------");
log.info("订阅者接收到的EmailAddress = "+emailRequest.getEmailAddress());
log.info("订阅者接收到的 Topic = "+emailRequest.getEmailTopic());
log.info("订阅者接收到的邮件的内容= "+emailRequest.getContent());
log.info("订阅者接收到的邮件发送的时间点= "+emailRequest.getCreateTime());
//这里目前网友可以根据当前系统时间 - emailRequest.getCreateTime() 比较，超过多长时间，可以丢弃此次订阅信息，不消费,直接return
log.info("------------------------------------------------------------------------");
log.info("订阅者开始订阅消息啦！");
log.info("开始发送邮件啦!");
//发邮件的代码逻辑(过程略),与本文介绍的内容没有太大联系！
//执行发送邮件的代码逻辑
log.info("发送邮件成功啦!");
log.info("订阅者结束订阅消息啦！");
log.info("结束发送邮件啦!");
log.info("End Send Email:" + emailRequest.getEmailAddress());
log.info("------------------------------------------------------------------------");
} catch(Exception ex) {
log.error("EmailMessageListener exception", ex);
}
}

public Executor getExecutor() {

// TODO Auto-generated method stub
return null;
}

}

----------------------------------

public class MessageSessionFactoryManager {
private static Log log = LogFactory.getLog(MessageSessionFactoryManager.class);
private static MessageSessionFactory sessionFactory = null;
private static String messagedir;
public static String getMessagedir() {
return messagedir;
}

private MessageSessionFactoryManager() {
}
public synchronized static MessageSessionFactory getSessionFactory() {
if(sessionFactory == null) {
init(true);
}
return sessionFactory;
}
public synchronized static MessageSessionFactory getSessionFactory(boolean isProducer) {
if(sessionFactory == null) {
init(isProducer);
}
return sessionFactory;
}
private static void init(boolean isProducer) {
try {
String confFile = "/metaq.ini";
InputStream in = EmailMessageProducerManager.class
.getResourceAsStream(confFile);

Properties conf = new Properties();

conf.load(in);

String zookeeper = conf.getProperty("zookeeper");

messagedir = conf.getProperty("messagedir");
final MetaClientConfig metaClientConfig = new MetaClientConfig();
final ZKConfig zkConfig = new ZKConfig();
zkConfig.zkConnect = zookeeper;
metaClientConfig.setZkConfig(zkConfig);
sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
//如果是生产者,则传入值为真;如果是消费者,则传入值为假。可以共用同一个sessionFactory
if(isProducer) {
//补偿机制，这里主要是针对如下两种情况
//发布者在onException中产生了异常，或者 result.isSuccess()返回值不成功，写会入到meaq.ini文件中对应的messagedir目录下
//然后由定时器每间隔30分钟扫描一次，扫到文件后，然后进行补偿机制进行重新由producer重新发布
Timer timer = new Timer();
//在2秒后执行此任务,每次间隔半小时扫描 D:\metaq\mmp\logs 目录下的异常文件，进行补偿机制发布消息 MesaageExceptionHandleTask（）实现方法很简单(由于字数限制不贴上去了)

timer.schedule(new MesaageExceptionHandleTask(), 2000, 1000*30*60);

}
} catch (Exception e) {
log.error(e);
throw new RuntimeException(e.getCause());
}
}
public static MessageProducer getMessageProducer() {
return getSessionFactory(true).createProducer();
}

}