- 浏览: 147295 次
- 性别:
- 来自: 北京
文章分类
最新评论
-
陈碧滔:
java获取工程目录路径 -
stevenjohn:
楼主能上传一下源代码吗?
ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息 -
指挥家:
讲的很好!
Foxmail6收发Web mail邮箱邮件实现原理 -
azhqiang:
谢谢你啊。呵呵
flex最全的表单验证
在P2P模型中,有下列概念:消息队列(Queue)、发送者(Sender)、接收者(Receiver)。每个消息都被发送到一个特定的队列,接收者从队列中获取消息。队列保留着消息,直到它们被消费或超时。
每个消息只有一个消费者(Consumer)(即一旦被消费,消息就不再在消息队列中)
发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息之后,不管接收者有没有正在运行,它不会影响到消息被发送到队列。
接收者在成功接收消息之后需向队列应答成功
如果你希望发送的每个消息都应该被成功处理的话,那么你需要P2P模型。
举例:
//注册消息监听器,当有消息发送过来的时候会调用onMessage方法(实现MessageListener 接口)
- import javax.ejb.ActivationConfigProperty;
- import javax.ejb.MessageDriven;
- import javax.jms.JMSException;
- import javax.jms.Message;
- import javax.jms.MessageListener;
- import javax.jms.TextMessage;
- @MessageDriven(activationConfig={
- @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Queue"),
- @ActivationConfigProperty(propertyName="destination", propertyValue="queue/myqueue")
- }
- )
- public class QueueMessageBean implements MessageListener {
- public void onMessage(Message msg) {
- //共有下面几种消息类型
- //1 Text
- //2 Map
- //3 Object
- //4 stream
- //5 byte
- TextMessage txtMsg = (TextMessage)msg;
- String s = "";
- try {
- s = txtMsg.getText();
- } catch (JMSException e) {
- e.printStackTrace();
- }
- System.out.println("QueueMessageBean接收到了消息:" + s);
- }
- }
- //客户端调用
- import javax.jms.Message;
- import javax.jms.MessageProducer;
- import javax.jms.Queue;
- import javax.jms.QueueConnection;
- import javax.jms.QueueConnectionFactory;
- import javax.jms.QueueSession;
- import javax.naming.InitialContext;
- public class Test {
- public static void main(String[] args) throws Exception {
- InitialContext ctx = new InitialContext();
- //获得QueueConnectionFactory对象
- QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");
- //创建QueueConnection对像
- QueueConnection connection = factory.createQueueConnection();
- //创建会话
- //arg1:与事物有关,true表示最后提交,false表示自动提交
- //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型
- QueueSession session = (QueueSession) connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
- //取得destination
- Queue queue = (Queue) ctx.lookup("queue/myqueue");
- //消息生产者
- MessageProducer sender = session.createProducer(queue);
- //定义消息
- Message msg = session.createTextMessage("消息来了");
- //发送消息
- sender.send(queue, msg);
- session.close();
- connection.close();
- }
- }
import javax.ejb.ActivationConfigProperty; import javax.ejb.MessageDriven; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; @MessageDriven(activationConfig={ @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Queue"), @ActivationConfigProperty(propertyName="destination", propertyValue="queue/myqueue") } ) public class QueueMessageBean implements MessageListener { public void onMessage(Message msg) { //共有下面几种消息类型 //1 Text //2 Map //3 Object //4 stream //5 byte TextMessage txtMsg = (TextMessage)msg; String s = ""; try { s = txtMsg.getText(); } catch (JMSException e) { e.printStackTrace(); } System.out.println("QueueMessageBean接收到了消息:" + s); } } //客户端调用 import javax.jms.Message; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.QueueConnection; import javax.jms.QueueConnectionFactory; import javax.jms.QueueSession; import javax.naming.InitialContext; public class Test { public static void main(String[] args) throws Exception { InitialContext ctx = new InitialContext(); //获得QueueConnectionFactory对象 QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("QueueConnectionFactory"); //创建QueueConnection对像 QueueConnection connection = factory.createQueueConnection(); //创建会话 //arg1:与事物有关,true表示最后提交,false表示自动提交 //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型 QueueSession session = (QueueSession) connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE); //取得destination Queue queue = (Queue) ctx.lookup("queue/myqueue"); //消息生产者 MessageProducer sender = session.createProducer(queue); //定义消息 Message msg = session.createTextMessage("消息来了"); //发送消息 sender.send(queue, msg); session.close(); connection.close(); } }
2、Pub/Sub模式
在Pub/Sub模型中,有下列概念: 主题(Topic)、发布者(Publisher)、订阅者(Subscriber)。客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。
每个消息可以有多个消费者
发布者和订阅者之间有时间上的依赖性。针对某个主题(Topic)的订阅者,它必须创建一个订阅之后,才能消费发布者的消息,而且,为了消费消息,订阅者必须保持运行的状态。
当然,为了缓和这种严格的时间相关性,JMS允许订阅者创建一个可持久化的订阅。这样,即使订阅者没有被激活(运行),它也能接收到发布者的消息。
如果你希望发送的消息可以不被做任何处理、或者被一个消费者处理、或者可以被多个消费者处理的话,那么可以采用Pub/Sub模型。
//注册消息监听器,当有消息发送过来的时候会调用onMessage方法(实现MessageListener 接口)
- import javax.ejb.ActivationConfigProperty;
- import javax.ejb.MessageDriven;
- import javax.jms.JMSException;
- import javax.jms.Message;
- import javax.jms.MessageListener;
- import javax.jms.TextMessage;
- @MessageDriven(activationConfig={
- @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Topic"),
- @ActivationConfigProperty(propertyName="destination", propertyValue="topic/myTopic")
- }
- )
- public class TopicMessageBean implements MessageListener {
- public void onMessage(Message msg) {
- //共有下面几种消息类型
- //1 Text
- //2 Map
- //3 Object
- //4 stream
- //5 byte
- TextMessage txtMsg = (TextMessage)msg;
- String s = "";
- try {
- s = txtMsg.getText();
- } catch (JMSException e) {
- e.printStackTrace();
- }
- System.out.println("TopicMessageBean接收到了消息:" + s);
- }
- }
- //客户端测试
- import javax.jms.MessageProducer;
- import javax.jms.Topic;
- import javax.jms.TopicConnection;
- import javax.jms.TopicConnectionFactory;
- import javax.jms.TopicSession;
- import javax.naming.InitialContext;
- public class Test {
- public static void main(String[] args) throws Exception {
- InitialContext ctx = new InitialContext();
- //获得QueueConnectionFactory对象
- TopicConnectionFactory factory = (TopicConnectionFactory) ctx.lookup("TopicConnectionFactory");
- //创建QueueConnection对像
- TopicConnection connection = factory.createTopicConnection();
- //创建会话
- //arg1:与事物有关,true表示最后提交,false表示自动提交
- //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型
- TopicSession session = (TopicSession) connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
- //取得destination
- Topic queue = (Topic) ctx.lookup("topic/myTopic");
- //消息生产者
- MessageProducer publisher = session.createProducer(queue);
- //定义消息
- Message msg = session.createTextMessage("消息来了");
- //发送消息
- publisher.send(queue, msg);
- session.close();
- connection.close();
- }
- }
import javax.ejb.ActivationConfigProperty; import javax.ejb.MessageDriven; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; @MessageDriven(activationConfig={ @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Topic"), @ActivationConfigProperty(propertyName="destination", propertyValue="topic/myTopic") } ) public class TopicMessageBean implements MessageListener { public void onMessage(Message msg) { //共有下面几种消息类型 //1 Text //2 Map //3 Object //4 stream //5 byte TextMessage txtMsg = (TextMessage)msg; String s = ""; try { s = txtMsg.getText(); } catch (JMSException e) { e.printStackTrace(); } System.out.println("TopicMessageBean接收到了消息:" + s); } } //客户端测试 import javax.jms.MessageProducer; import javax.jms.Topic; import javax.jms.TopicConnection; import javax.jms.TopicConnectionFactory; import javax.jms.TopicSession; import javax.naming.InitialContext; public class Test { public static void main(String[] args) throws Exception { InitialContext ctx = new InitialContext(); //获得QueueConnectionFactory对象 TopicConnectionFactory factory = (TopicConnectionFactory) ctx.lookup("TopicConnectionFactory"); //创建QueueConnection对像 TopicConnection connection = factory.createTopicConnection(); //创建会话 //arg1:与事物有关,true表示最后提交,false表示自动提交 //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型 TopicSession session = (TopicSession) connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE); //取得destination Topic queue = (Topic) ctx.lookup("topic/myTopic"); //消息生产者 MessageProducer publisher = session.createProducer(queue); //定义消息 Message msg = session.createTextMessage("消息来了"); //发送消息 publisher.send(queue, msg); session.close(); connection.close(); } }
二种模型的实现结果:对于p2p模型的每个消息只能有一个消费者 如果我们定义二个消息接受者的Bean那么只能有一端会接收到消息。当你把部署在Jboss中的消息接收Bean去掉以后,然后发送消息 此时消息在队列中,一旦你重新部署他会立刻就接收到刚刚发送的消息所以它没有时间的依赖性, pub/sub模型可以有多个消费者 在这个模型中如果我们定义多个接收消息的Bean当我们在客户端发送消息的时候二个bean都会接收到消息,所以他有多个消费者 但是如果你把Jboss部署中的消息接收bean去掉之后,发送消息。然后在重新部署,那么消息也无法接收到,所以说他有时间的依赖性。
//代码中几个概念的理解
Connection Factory
创建Connection对象的工厂,针对两种不同的JMS消息模型,分别有QueueConnectionFactory和TopicConnectionFactory两种。可以通过JNDI来查找ConnectionFactory对象。
Destination
Destination的意思是消息生产者的消息发送目标或者说消息消费者的消息来源。对于消息生产者来说,它的Destination是某个队列(Queue)或某个主题(Topic);对于消息消费者来说,它的Destination也是某个队列或主题(即消息来源)。
所以,Destination实际上就是两种类型的对象:Queue、Topic。
可以通过JNDI来查找Destination。
Connection:
Connection表示在客户端和JMS系统之间建立的链接(对TCP/IP socket的包装)。Connection可以产生一个或多个Session。跟ConnectionFactory一样,Connection也有两种类型:QueueConnection和TopicConnection。
Session:
Session是我们操作消息的接口。可以通过session创建生产者、消费者、消息等。Session提供了事务的功能。当我们需要使用session发送/接收多个消息时,可以将这些发送/接收动作放到一个事务中。同样,也分QueueSession和TopicSession。
消息生产者:
消息生产者由Session创建,并用于将消息发送到Destination。同样,消息生产者分两种类型:QueueSender和TopicPublisher。可以调用消息生产者的方法(send或publish方法)发送消息!
消息消费者:
消息消费者由Session创建,用于接收被发送到Destination的消息。两种类型:QueueReceiver和TopicSubscriber。可分别通过session的createReceiver(Queue)或createSubscriber(Topic)来创建。当然,也可以通过session的createDurableSubscriber方法来创建持久化的订阅者。
MessageListener:
消息监听器。如果注册了消息监听器,一旦消息到达,将自动调用监听器的onMessage方法。EJB中的MDB(Message-Driven Bean)就是一种MessageListener。
MDB介绍:
对客户端来说,message-driven bean就是异步消息的消费者。当消息到达之后,由容器负责调用MDB。客户端发送消息到destination,MDB作为一个MessageListener接收消息。
发表评论
-
Weblogic下创建JMS消息服务
2010-11-08 14:29 1842转载:http://yunzhongxia.iteye.com ... -
Spring整合Blazeds实现Weblogic JMS消息服务
2010-11-08 14:27 1489转载:http://yunzhongxia.iteye.com ... -
Spring整合Blazeds实现ActiveMQ JMS消息服务
2010-11-08 14:18 1782转载:http://yunzhongxia.iteye.com ... -
Blazeds+JMS(ActiveMQ)+Spring实现消息
2010-11-08 14:16 1844转载:http://yunzhongxia.iteye.com ... -
Jms之ActiveMQ学习笔记
2010-11-08 10:48 2729ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4 ... -
ActiveMQ5.0实战二: 基本配置
2010-11-02 10:04 1606简介 上一篇http://www.iteye.com/top ... -
ActiveMQ5.0实战一: 安装配置ActiveMQ5.0
2010-11-02 09:59 989简介 ActiveMQ 是开源的JMS实现,Geroni ... -
ActiveMQ文档学习
2010-11-01 13:59 2156研究ActiveMQ 的文档1. 背景当前,CORBA、DCO ...
相关推荐
Java消息服务支持两种消息模型:Point-to-Point消息(P2P)和发布订阅消息(Publish Subscribe messaging,简称Pub/Sub)。JMS规范并不要求供应商同时支持这两种消息模型,但开发者应该熟悉这两种消息模型的优势与缺点...
Java消息服务支持两种消息模型:Point-to-Point消息(P2P)和发布订阅消息(Publish Subscribe messaging,简称Pub/Sub)。JMS规范并不要求供应商同时支持这两种消息模型,但开发者应该熟悉这两种消息模型的优势与缺点...
JMS是Java的消息服务,JMS的客户端之间可以通过JMS服务进行异步的消息传输。JMS支持两种消息模型:Point-to-Point(P2P)和Publish/Subscribe(Pub/Sub),即点对点和发布订阅模型。
sudo docker run --volume=/:/rootfs:ro --volume=/var/run:/var/run:ro --volume=/sys:/sys:ro --volume=/var/lib/docker/:/var/lib/docker:ro --volume=/dev/disk/:/dev/disk:ro --publish=8080:8080 --detach=...
基于OpenFlow的发布订阅系统数据分发模块的研究与实现,章洋,陈天宇,在现今的网络中,所有网络协议只提供
发布/订阅(publish/subscribe,简称 pub/sub)。 这两种消息传递模型非常相似,但有以下区别: PTP 消息传递模型规定了一条消息只能传递给一个接收方。 采用javax.jms.Queue 表示。Spring配置类型destination-type=...
--volume=/:/rootfs:ro \ --volume=/var/run:/var/run:ro \ --volume=/sys:/sys:ro \ --volume=/var/lib/docker/:/var/lib/docker:ro \ --volume=/dev/disk/:/dev/disk:ro \ --publish=8080:8080 \ --detach=...
--publish 1443:443 --publish 18080:80 \ --restart always \ --volume /root/docker/gitlab/config:/etc/gitlab \ --volume /root/docker/gitlab/logs:/var/log/gitlab \ --volume /root/docker/gitlab/...
11) — an introduction to the Persistent Publish/Subscribe service, and how to run it • Objects and their attributes (p. 13) — a description of the PPS service's objects and their attributes • ...
Publish/Subscribe service, and how to run it • Objects and their attributes (p. 13) — a description of the PPS service's objects and their attributes • Persistence (p. 19) — how PPS manages ...
Java Message Service(JMS)是SUN提出的旨在统一各种MOM(Message-Oriented Middleware )系统接口的规范,它包含点对点(Point to Point,PTP)和发布/订阅(Publish/Subscribe,pub/sub)两种消息模型,提供可靠...
一种基于发布订阅系统的实时可靠分发方案,章洋,肖丹,基于主题的发布订阅系统已在现代物联网中广泛应用,但很多已实现的系统因为没有考虑一些实际因素(如主题之间的关系、操作限制等
* HTML 格式(在线中文) http://download.java.net/jdk/jdk-api-localizations/jdk-api-zh-cn/publish/1.6.0/html/zh_CN/api/index.html * zip 格式(中文) ...
基于SDN的发布/订阅系统中的路由计算,王双锦,章洋,本文提供一种基于SDN的发布订阅系统的基础架构和路由解决方案。通过研究应用OpenFlow协议的SDN网络对发布/订阅系统架构的传输效率和路
PUB_ROOT=/home/chenlong/Publish === 使用方法 ===: #./apply.sh blog /home/chenlong/patch_sh/list20100520.txt 例如 -bash-3.2$ ./apply.sh blog /home/chenlong/patch_sh/list20100520.txt /var/...
NULL 博文链接:https://kingxss.iteye.com/blog/1420264
CMain展示了Server和Client端的Util用法,以java工程运行CMain可以进行测试,可以选择测试publish(发布)或subscribe(订阅), 可使用Mosquitto命令行配合进行测试。 使用前配置properties文件中的brorkerServer的...
前端开源库-require-npm4-to-publishrequire-npm4-to-publish,一个模块,可以很容易地检测我们是否在为发布操作运行npm4。