`
XinTeng2012
  • 浏览: 94749 次
社区版块
存档分类
最新评论

Jms两种message传输方式Topic和Queue的比较

 
阅读更多

/**

* 转载请注明作者longdick http://longdick.iteye.com

*

*/

Jms规范里的两种message传输方式Topic和Queue,两者的对比如下表():


Topic Queue
概要 Publish Subscribe messaging 发布订阅消息 Point-to-Point 点对点
有无状态 topic数据默认不落地,是无状态的。

Queue数据默认会在mq服务器上以文件形式保存,比如Active MQ一般保存在$AMQ_HOME\data\kr-store\data下面。也可以配置成DB存储。

完整性保障 并不保证publisher发布的每条数据,Subscriber都能接受到。 Queue保证每条数据都能被receiver接收。
消息是否会丢失 一般来说publisher发布消息到某一个topic时,只有正在监听该topic地址的sub能够接收到消息;如果没有sub在监听,该topic就丢失了。 Sender发送消息到目标Queue,receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取,也不会丢失。
消息发布接收策略 一对多的消息发布接收策略,监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器 一对一的消息发布接收策略,一个sender发送的消息,只能有一个receiver接收。receiver接收完后,通知mq服务器已接收,mq服务器对queue里的消息采取删除或其他操作。

以下是符合Jms1.1规范的使用Queue传输消息的代码,使用Active mq作为Jms的实现,想要更clean的代码,可以考虑将Amq的实现DI:

Java代码 收藏代码
  1. publicvoidtestMySend()throwsJMSException{
  2. Connectionconnection=newActiveMQConnectionFactory("admin","admin","tcp://localhost:61616").createConnection();
  3. Sessionsession=null;
  4. Queuequeue=null;
  5. MessageProducerproducer=null;
  6. session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
  7. queue=session.createQueue("myTest");
  8. producer=session.createProducer(queue);
  9. Messagemsg=session.createTextMessage("hello");
  10. producer.send(msg);
  11. producer.close();
  12. session.close();
  13. connection.close();
  14. }
  15. publicvoidtestMyReceive()throwsJMSException{
  16. Connectionconnection=newActiveMQConnectionFactory("admin","admin","tcp://localhost:61616").createConnection();
  17. Sessionsession=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
  18. Queuequeue=session.createQueue("myTest");
  19. MessageConsumerconsumer=session.createConsumer(queue);
  20. consumer.setMessageListener(newMyListener());
  21. connection.start();
  22. consumer.close();
  23. session.close();
  24. connection.close();
  25. }

消费端需要设定一个MessageListener:

Java代码 收藏代码
  1. privateclassMyListenerimplementsMessageListener{
  2. publicvoidonMessage(Messagemessage){
  3. System.out.println("msgstart!----------------");
  4. try{
  5. System.out.println(""+((TextMessage)message).getText());
  6. }catch(JMSExceptione){
  7. e.printStackTrace();
  8. }
  9. System.out.println("msgend!----------------");
  10. }
  11. }

以下是符合Jms1.1规范的使用Topic 发送消息的代码:

Java代码 收藏代码
  1. publicvoidtestMyTopicPublisher()throwsJMSException{
  2. Connectionconnection=newActiveMQConnectionFactory("admin","admin","tcp://localhost:61616").createConnection();
  3. Sessionsession=null;
  4. Topictopic=null;
  5. MessageProducerproducer=null;
  6. session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
  7. topic=session.createTopic("myTopicTest");
  8. producer=session.createProducer(topic);
  9. producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  10. Messagemsg=session.createTextMessage("helloandwhateverusay");
  11. producer.send(msg);
  12. producer.close();
  13. session.close();
  14. connection.close();
  15. }

以下是符合Jms1.1规范的使用Topic 接收消息的代码:

Java代码 收藏代码
  1. publicstaticvoidtestMyTopicConsumer()throwsJMSException,InterruptedException{
  2. Connectionconnection=newActiveMQConnectionFactory("admin","admin","tcp://localhost:61616").createConnection();
  3. Sessionsession=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);
  4. Topictopic=session.createTopic("myTopicTest");
  5. MessageConsumerconsumer=session.createConsumer(topic);
  6. consumer.setMessageListener(newMyListener());
  7. connection.start();
  8. }

jms在创建Session时可以有两个参数,第一个参数是是否使用事务,第二个参数是消费者向发送者确认消息已经接收的方式:

Java代码 收藏代码
  1. Sessionsession=connection.createSession(false,Session.CLIENT_ACKNOWLEDGE);

确认消息的方式有如下三种:

AUTO_ACKNOWLEDGE(自动通知)

CLIENT_ACKNOWLEDGE(客户端自行决定通知时机)

DUPS_OK_ACKNOWLEDGE(延时//批量通知)

如果使用的是 客户端自行决定通知时机 方式,那么需要在MessageListener里显式调用message.acknowledge()来通知服务器。服务器接收到通知后采取相应的操作。

分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics