`

主题:JMS的两种消息模型(Point-to-Point(P2P)和Publish/Subscribe(Pub/Sub))应用举例

    博客分类:
  • JMS
阅读更多
1、P2P模型
在P2P模型中,有下列概念:消息队列(Queue)、发送者(Sender)、接收者(Receiver)。每个消息都被发送到一个特定的队列,接收者从队列中获取消息。队列保留着消息,直到它们被消费或超时。
 每个消息只有一个消费者(Consumer)(即一旦被消费,消息就不再在消息队列中)
 发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息之后,不管接收者有没有正在运行,它不会影响到消息被发送到队列。
 接收者在成功接收消息之后需向队列应答成功
如果你希望发送的每个消息都应该被成功处理的话,那么你需要P2P模型。
举例:
//注册消息监听器,当有消息发送过来的时候会调用onMessage方法(实现MessageListener 接口)
Java代码 复制代码
  1. import javax.ejb.ActivationConfigProperty;   
  2. import javax.ejb.MessageDriven;   
  3. import javax.jms.JMSException;   
  4. import javax.jms.Message;   
  5. import javax.jms.MessageListener;   
  6. import javax.jms.TextMessage;   
  7.   
  8. @MessageDriven(activationConfig={   
  9.             @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Queue"),   
  10.             @ActivationConfigProperty(propertyName="destination", propertyValue="queue/myqueue")   
  11.     }   
  12. )   
  13. public class QueueMessageBean implements MessageListener {   
  14.   
  15.     public void onMessage(Message msg) {   
  16.         //共有下面几种消息类型   
  17.         //1 Text   
  18.         //2 Map   
  19.         //3 Object   
  20.         //4 stream   
  21.         //5 byte   
  22.         TextMessage txtMsg = (TextMessage)msg;   
  23.         String s = "";   
  24.         try {   
  25.             s = txtMsg.getText();   
  26.         } catch (JMSException e) {   
  27.             e.printStackTrace();   
  28.         }   
  29.         System.out.println("QueueMessageBean接收到了消息:" + s);   
  30.     }   
  31. }   
  32. //客户端调用   
  33. import javax.jms.Message;   
  34. import javax.jms.MessageProducer;   
  35. import javax.jms.Queue;   
  36. import javax.jms.QueueConnection;   
  37. import javax.jms.QueueConnectionFactory;   
  38. import javax.jms.QueueSession;   
  39. import javax.naming.InitialContext;   
  40.   
  41.   
  42. public class Test {   
  43.     public static void main(String[] args) throws Exception {   
  44.     InitialContext ctx = new InitialContext();   
  45.     //获得QueueConnectionFactory对象   
  46.     QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");   
  47.     //创建QueueConnection对像    
  48.     QueueConnection connection = factory.createQueueConnection();   
  49.     //创建会话   
  50.     //arg1:与事物有关,true表示最后提交,false表示自动提交   
  51.     //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型   
  52.     QueueSession session = (QueueSession) connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);   
  53.     //取得destination   
  54.     Queue queue = (Queue) ctx.lookup("queue/myqueue");   
  55.     //消息生产者   
  56.     MessageProducer sender = session.createProducer(queue);   
  57.     //定义消息   
  58.     Message msg = session.createTextMessage("消息来了");   
  59.     //发送消息   
  60.     sender.send(queue, msg);   
  61.     session.close();   
  62.     connection.close();   
  63.            
  64.     }   
  65. }  
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 接口)

Java代码 复制代码
  1. import javax.ejb.ActivationConfigProperty;   
  2. import javax.ejb.MessageDriven;   
  3. import javax.jms.JMSException;   
  4. import javax.jms.Message;   
  5. import javax.jms.MessageListener;   
  6. import javax.jms.TextMessage;   
  7.   
  8. @MessageDriven(activationConfig={   
  9.             @ActivationConfigProperty(propertyName="destinationType",propertyValue="javax.jms.Topic"),   
  10.             @ActivationConfigProperty(propertyName="destination", propertyValue="topic/myTopic")   
  11.     }   
  12. )   
  13. public class TopicMessageBean implements MessageListener {   
  14.   
  15.     public void onMessage(Message msg) {   
  16.         //共有下面几种消息类型   
  17.         //1 Text   
  18.         //2 Map   
  19.         //3 Object   
  20.         //4 stream   
  21.         //5 byte   
  22.         TextMessage txtMsg = (TextMessage)msg;   
  23.         String s = "";   
  24.         try {   
  25.             s = txtMsg.getText();   
  26.         } catch (JMSException e) {   
  27.             e.printStackTrace();   
  28.         }   
  29.         System.out.println("TopicMessageBean接收到了消息:" + s);   
  30.     }   
  31. }   
  32.   
  33. //客户端测试   
  34. import javax.jms.MessageProducer;   
  35. import javax.jms.Topic;   
  36. import javax.jms.TopicConnection;   
  37. import javax.jms.TopicConnectionFactory;   
  38. import javax.jms.TopicSession;   
  39. import javax.naming.InitialContext;   
  40.   
  41.   
  42. public class Test {   
  43.     public static void main(String[] args) throws Exception {   
  44.     InitialContext ctx = new InitialContext();   
  45.     //获得QueueConnectionFactory对象   
  46.     TopicConnectionFactory factory = (TopicConnectionFactory) ctx.lookup("TopicConnectionFactory");   
  47.     //创建QueueConnection对像    
  48.     TopicConnection connection = factory.createTopicConnection();   
  49.     //创建会话   
  50.     //arg1:与事物有关,true表示最后提交,false表示自动提交   
  51.     //arg2:表示消息向中间件发送确认通知,这里采用的是自动通知的类型   
  52.     TopicSession session = (TopicSession) connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);   
  53.     //取得destination   
  54.     Topic queue = (Topic) ctx.lookup("topic/myTopic");   
  55.     //消息生产者   
  56.     MessageProducer publisher = session.createProducer(queue);   
  57.     //定义消息   
  58.     Message msg = session.createTextMessage("消息来了");   
  59.     //发送消息   
  60.     publisher.send(queue, msg);   
  61.     session.close();   
  62.     connection.close();   
  63.            
  64.     }   
  65. }  
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接收消息。

分享到:
评论

相关推荐

    JMS sub/pub实现聊天系统

    在JMS中,有两种消息传递模型:点对点(Point-to-Point)和发布/订阅(Publish/Subscribe)。聊天系统的特性更适合使用发布/订阅模型,因为它允许多个订阅者接收来自同一消息源的消息,类似于广播。在这个模型中,...

    JMS简明教程(Java消息服务)

    Java消息服务支持两种消息模型:Point-to-Point消息(P2P)和发布订阅消息(Publish Subscribe messaging,简称Pub/Sub)。JMS规范并不要求供应商同时支持这两种消息模型,但开发者应该熟悉这两种消息模型的优势与缺点...

    javax.jms-1.1.jar

    2. **消息模型**:JMS支持两种消息模型——点对点(Point-to-Point, PTP)和发布/订阅(Publish/Subscribe, Pub/Sub)。在点对点模型中,消息从一个队列(Queue)中发送到另一个队列,每个消息仅被一个消费者接收。...

    ActiveMQ的处理模式:PTP与PUB/SUB

    在ActiveMQ中,有两种主要的消息处理模式:点对点(Point-to-Point,简称PTP)和发布/订阅(Publish/Subscribe,简称PUB/SUB)。本文将深入探讨这两种模式及其在SpringBoot应用中的实现。 首先,点对点(PTP)模式...

    jms-1.1.jar

    JMS支持两种主要的消息模型:点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe)。 点对点模型中,消息由一个生产者发送到一个队列,然后由一个或多个消费者接收,但每个消息只能被一个消费者消费。...

    jms-1_1-fr-apidocs.zip

    1. **消息模型**:JMS支持两种消息模型——点对点(Point-to-Point,PTP)和发布/订阅(Publish/Subscribe,Pub/Sub)。在点对点模型中,消息由一个生产者发送到一个队列,然后由一个消费者接收;而在发布/订阅模型...

    JMS简明教程创建消息 -> 发送消息 -> 接收消息 -> 读取消息

    Java消息服务支持两种消息模型:Point-to-Point消息(P2P)和发布订阅消息(Publish Subscribe messaging,简称Pub/Sub)。JMS规范并不要求供应商同时支持这两种消息模型,但开发者应该熟悉这两种消息模型的优势与缺点...

    JMS消息模型 JMS学习.doc

    3. **JMS Domains**:JMS有两种主要的消息模型,即点对点(Point-to-Point, PTP)和发布/订阅(Publish/Subscribe, Pub/Sub)模式。 - **Point-to-Point (PTP)模式**:在这种模型中,消息从一个生产者发送到一个...

    消息中间件和JMS消息服务.pdf

    JMS支持两种基本的消息模型:点对点(Point-to-Point, PTP)和发布/订阅(Publish/Subscribe, Pub/Sub)。这两种模型分别对应于不同的应用场景和需求。 1. **点对点模型**:在此模型中,消息被发送到特定的队列,每个...

    J2EE EJB规范文档

    JMS是Java的消息服务,JMS的客户端之间可以通过JMS服务进行异步的消息传输。JMS支持两种消息模型:Point-to-Point(P2P)和Publish/Subscribe(Pub/Sub),即点对点和发布订阅模型。

    jboss-jms包

    - **消息模型**:JMS支持两种消息模型,点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe,Pub/Sub)。 - P2P模型中,消息从一个生产者发送到一个队列,然后由一个或多个消费者消费。消息一旦被消费...

    JMS之ActiveMQ 点对点+发布/订阅

    在JMS中,消息传递有两种基本模型:点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe,Pub/Sub)。 **点对点(Point-to-Point,P2P)模型** 点对点模型是基于队列(Queue)的通信方式。在这个模型中,...

    jms简介.pdf

    JMS 规范支持两种主要的消息模型:点对点 (Point-to-Point, P2P) 和发布/订阅 (Publish/Subscribe, Pub/Sub)。 ##### 3.1 点对点消息模型 - **特点**:P2P 模型确保每个消息至少被一个消费者接收且仅被接收一次。 ...

    sun-jms权威指南

    - **编程域**:分为点对点 (Point-to-Point, P2P) 和发布/订阅 (Publish/Subscribe, Pub/Sub) 模型。 - **可靠消息传送**: - **确认/事务**:确保消息被正确处理的机制。 - **持久性存储器**:用于存储未被消费...

    jms-test.zip

    在JMS中,有两种主要的消息模式:点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe,Pub/Sub)模式。这两种模式各有特点,适用于不同的场景。 1. 点对点(Point-to-Point,P2P)模式: 在P2P模式下,...

    jms-1.1+jmxri-1.2.1+jmxtools-1.2.1.zip

    在JMS-1.1规范中,主要有两种类型的消息模型:点对点(Point-to-Point,P2P)和发布/订阅(Publish/Subscribe,Pub/Sub)。点对点模型通过队列(Queue)实现,每个消息只有一个消费者;而发布/订阅模型通过主题...

    JMS 教程 - 消息队列、消息服务

    JMS支持两种主要的消息传递模型:发布/订阅(Publish and Subscribe)和点对点(Point-to-Point)。这两种模型分别对应于不同的应用场景。 **发布/订阅(Publish and Subscribe)** 模型适用于一对多的通信场景。在...

    基于JMS的消息中间件的实现-论文.zip

    JMS提供两种主要的消息模型:点对点(Point-to-Point,PTP)和发布/订阅(Publish/Subscribe,Pub/Sub)。在PTP模型中,消息从一个生产者发送到一个队列,然后由一个消费者接收;而在Pub/Sub模型中,消息发布到一个...

    jms学习指导(weblogic下的)

    - **发布/订阅(Publish/Subscribe, Pub/Sub)**:在这个模型中,一个消息可以被多个订阅者接收。发布者发送消息到主题,而订阅者可以订阅这些主题以接收消息。 **JMS基本功能** 1. **消息组件**:消息由三部分...

    Spring 实现远程访问详解——jms和activemq

    发布/订阅(publish/subscribe,简称 pub/sub)。 这两种消息传递模型非常相似,但有以下区别: PTP 消息传递模型规定了一条消息只能传递给一个接收方。 采用javax.jms.Queue 表示。Spring配置类型destination-type=...

Global site tag (gtag.js) - Google Analytics