分布式消息队列中间件系列研究之阿堂教程(进阶篇)-阿里云开发者社区

开发者社区> 开发与运维> 正文
登录阅读全文

分布式消息队列中间件系列研究之阿堂教程(进阶篇)

简介:

 上篇文章,阿堂和大家分享了《分布式消息队列中间件系列研究之阿堂教程(基础篇-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("heyitang@qq.com", 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();
}

}



本文转自 www19 51CTO博客,原文链接:http://blog.51cto.com/doujh/1715275,如需转载请自行联系原作者

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享: