JMS的两种消息模型(Point-to-Point(P2P)和Publish/Subscribe(Pub/Sub))应用举例
http://wuzhaohuixy-qq-com.iteye.com/blog/908395
在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接收消息。
JMS的两种消息模型(Point-to-Point(P2P)和Publish/Subscribe(Pub/Sub))应用举例相关推荐
- day72 JavaWeb框架阶段——RabbitMQ消息队列【了解常见的MQ产品,了解RabbitMQ的5种消息模型,会使用Spring AMQP】
文章目录 0.学习目标 1.RabbitMQ 1.1.搜索与商品服务的问题 1.2.消息队列(MQ) 1.2.1.什么是消息队列 1.2.2.AMQP和JMS 1.2.3.常见MQ产品 1.2.4.R ...
- 以两种异步模型应用案例,深度解析Future接口
摘要:本文以实际案例的形式分析了两种异步模型,并从源码角度深度解析Future接口和FutureTask类. 本文分享自华为云社区<[精通高并发系列]两种异步模型与深度解析Future接口(一) ...
- ActiveMQ的两种消息模式及实现代码
消息列队有两种消息模式,一种是点对点的消息模式,还有一种就是订阅的模式. (1).点对点方式(point-to-point) 点对点的消息发送方式主要建立在 Message Queue,Sender, ...
- Redis的两种消息模式
Redis的两种消息模式 队列模式 发布订阅模式 队列模式 队列模式下每个消费者可以同时从多个服务器读取消息,但是每个消息只能被一个消费者读取. 在队列模式下其实每次插入的数据都是载入在最前面的,而先 ...
- 深度解析两种信用评估模型
"大数据"概念的最早出现,是从2012年2月份纽约时报一篇文章开始的.到目前为止,在大数据领域当中的投资已经越来越热,该领域企业越来越多.大数据在美国金融当中最直接的场景,主要运用 ...
- SpringBoot整合RabbitMQ 实现五种消息模型
目录 SpringBoot中使用RabbitMQ 搭建初始环境 引入依赖 配置配置文件 测试类 注入 rabbitTemplate 消息队列RabbitMQ之五种消息模型 第一种直连模型使用 开发生产 ...
- DIV+CSS两种盒子模型(W3C盒子与IE盒子)
在辨析两种盒子模型之前.先简单说明一下什么叫盒子模型. 原理: 先说说我们在网页设计中常听的属性名:内容(content).填充(padding).边框(border).边界(margin), CSS ...
- Python实现RabbitMQ中6种消息模型(转)
RabbitMQ与Redis对比 RabbitMQ是一种比较流行的消息中间件,之前我一直使用redis作为消息中间件,但是生产环境比较推荐RabbitMQ来替代Redis,所以我去查询了一些Rab ...
- RabbitMQ之五种消息模型
首先什么是MQ MQ全称是Message Queue,即消息对列!消息队列是典型的:生产者.消费者模型.生产者不断向消息队列中生产消息,消费者不断的从队列中获取消息.因为消息的生产和消费都是异步的,而 ...
最新文章
- C++ cin 实现循环读入
- python求矩形面积_关于python:创建矩形类
- Docker安装禅道项目管理平台zendao-v11.6.x
- leetcode —— 39. 组合总和
- Vue使用Vditor编辑器
- 阿里云镜像加速Docker
- PowerAI DDL
- 用正则彻底去除HTML\CSS\script代码
- 2022最新版sci和ssci双检索期刊,详细期刊目录值得了解!
- 2021年最新程序员培训机构排名,学习前避坑必看
- 基于单片机的智能电风扇设计
- 先旋转x在旋转y 在旋转z_旋转Kubernetes中的秘密
- 依赖缺失问题 CS0246	未能找到类型或命名空间名“SqlConnection”(是否缺少 using 指令或程序集引用?) CS1069 未能在命名空间“System.Data.SqlClient
- el-popover 最小宽度 min-width
- Android 利用广播实现短信的自动转发
- 圣杯布局和双飞翼布局详解
- 7-13 非常弹的球 (30 分)
- 五家共井c语言,五家共井—最早的不定方程
- 关于手机号码合法性判断
- ClassCastException: $Proxy0 cannot be cast to (com......) 的解决办法
热门文章
- 采用8种相位,每种相位各有两种幅度的QAM调制方法,在1200Baud的信号传输速率下能达到的网数据传输速率为( )
- LeetCode 多线程 1116. 打印零与奇偶数
- 2014年第五届蓝桥杯C/C++ A组国赛 —— 第四题:排列序数
- 【机器视觉】 set_fuzzy_measure_norm_pair算子
- 【机器视觉】 dev_show_tool算子
- 【Linux】一步一步学Linux——dhclient命令(156)
- [Qt教程] 第23篇 数据库(三)利用QSqlQuery类执行SQL语句
- C++学习笔记-----在一个构造函数中调用另一个构造函数
- dhrystone测试结果_RTThread软件包可以对MCU进行性能测试,跑一个试试!
- LVS(7)——NAT实践