Java消息队列JMS实现原理解析

一、什么是JMS

JMS即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持(百度百科给出的概述)。我们可以简单的理解:两个应用程序之间需要进行通信,我们使用一个JMS服务,进行中间的转发,通过JMS 的使用,我们可以解除两个程序之间的耦合。

二、为什么需要JMS

在JAVA中,如果两个应用程序之间对各自都不了解,甚至这两个程序可能部署在不同的大洲上,那么它们之间如何发送消息呢?举个例子,一个应用程序A部署在印度,另一个应用程序部署在美国,然后每当A触发某件事后,B想从A获取一些更新信息。当然,也有可能不止一个B对A的更新信息感兴趣,可能会有N个类似B的应用程序想从A中获取更新的信息。

在这种情况下,JAVA提供了最佳的解决方案-JMS,完美解决了上面讨论的问题。

JMS同样适用于基于事件的应用程序,如聊天服务,它需要一种发布事件机制向所有与服务器连接的客户端发送消息。JMS与RMI不同,发送消息的时候,接收者不需要在线。服务器发送了消息,然后就不管了;等到客户端上线的时候,能保证接收到服务器发送的消息。这是一个很强大的解决方案,能处理当今世界很多普遍问题。

三、JMS的优势

1、异步

JMS天生就是异步的,客户端获取消息的时候,不需要主动发送请求,消息会自动发送给可用的客户端。

2、可靠

JMS保证消息只会递送一次。大家都遇到过重复创建消息问题,而JMS能帮你避免该问题。

四、JMS的消息模型

JMS具有两种通信模式:

1、Point-to-Point Messaging Domain (点对点)

2、Publish/Subscribe Messaging Domain (发布/订阅模式)

在JMS API出现之前,大部分产品使用“点对点”和“发布/订阅”中的任一方式来进行消息通讯。JMS定义了这两种消息发送模型的规范,它们相互独立。任何JMS的提供者可以实现其中的一种或两种模型,这是它们自己的选择。JMS规范提供了通用接口保证我们基于JMS API编写的程序适用于任何一种模型。

1、Point-to-Point Messaging Domain(点对点通信模型)

点对点的模式主要建立在一个队列上面,当连接一个列队的时候,发送端不需要知道接收端是否正在接收,可以直接向ActiveMQ发送消息,发送的消息,将会先进入队列中,如果有接收端在监听,则会发向接收端,如果没有接收端接收,则会保存在activemq服务器,直到接收端接收消息,点对点的消息模式可以有多个发送端,多个接收端,但是一条消息,只会被一个接收端给接收到,哪个接收端先连上ActiveMQ,则会先接收到,而后来的接收端则接收不到那条消息。

在点对点消息传送模型中,应用程序由消息队列,发送者,接收者组成。每一个消息发送给一个特殊的消息队列,该队列保存了所有发送给它的消息(除了被接收者消费掉的和过期的消息)。

(1)模式图:

(2)特点:

每个消息只有一个消费者;

发送者和接收者在时间上是没有时间的约束,也就是说发送者在发送完消息之后,不管接收者有没有接受消息,都不会影响发送方发送消息到消息队列中;

发送方不管是否在发送消息,接收方都可以从消息队列中取到消息;

接收方在接收完消息之后,需要向消息队列应答成功。

2、Publish/Subscribe Messaging Domain(发布/订阅通信模型)

订阅/发布模式,同样可以有着多个发送端与多个接收端,但是接收端与发送端存在时间上的依赖,就是如果发送端发送消息的时候,接收端并没有监听消息,那么ActiveMQ将不会保存消息,将会认为消息已经发送,换一种说法,就是发送端发送消息的时候,接收端不在线,是接收不到消息的,哪怕以后监听消息,同样也是接收不到的。这个模式还有一个特点,那就是,发送端发送的消息,将会被所有的接收端给接收到,不类似点对点,一条消息只会被一个接收端给接收到。

在发布/订阅消息模型中,发布者发布一个消息,该消息通过topic传递给所有的客户端。该模式下,发布者与订阅者都是匿名的,即发布者与订阅者都不知道对方是谁。并且可以动态的发布与订阅Topic。Topic主要用于保存和传递消息,且会一直保存消息直到消息被传递给客户端。

(1)模式图:

(2)特点:

一个消息可以传递个多个订阅者(即:一个消息可以有多个接收方);

发布者与订阅者具有时间约束,针对某个主题(Topic)的订阅者,它必须创建一个订阅者之后,才能消费发布者的消息,而且为了消费消息,订阅者必须保持运行的状态;

为了缓和这样严格的时间相关性,JMS允许订阅者创建一个可持久化的订阅。这样,即使订阅者没有被激活(运行),它也能接收到发布者的消息。

五、JMS接收消息

在JMS中,消息的接收可以使用以下两种方式:

(1)同步

使用同步方式接收消息的话,消息订阅者调用receive()方法。在receive()中,消息未到达或在到达指定时间之前,方法会阻塞,直到消息可用。

(2)异步

使用异步方式接收消息的话,消息订阅者需注册一个消息监听者,类似于事件监听器,只要消息到达,JMS服务提供者会通过调用监听器的onMessage()递送消息。

六、JMS编程模型

(1)管理对象(Administered objects)-连接工厂(Connection Factories)和目的地(Destination)

(2)连接对象(Connections)

(3)会话(Sessions)

(4)消息生产者(Message Producers)

(5)消息消费者(Message Consumers)

(6)消息监听者(Message Listeners

(1)JMS管理对象

管理对象(Administered objects)是预先配置的JMS对象,由系统管理员为使用JMS的客户端创建,主要有两个被管理的对象:

连接工厂(ConnectionFactory)目的地(Destination)

这两个管理对象由JMS系统管理员通过使用Application Server管理控制台创建,存储在应用程序服务器的JNDI名字空间或JNDI注册表。

① 连接工厂(ConnectionFactory)

创建Connection对象的工厂,针对两种不同的jms消息模型,分别有QueueConnectionFactory和TopicConnectionFactory两种。可以通过JNDI来查找ConnectionFactory对象。客户端使用一个连接工厂对象连接到JMS服务提供者,它创建了JMS服务提供者和客户端之间的连接。JMS客户端(如发送者或接受者)会在JNDI名字空间中搜索并获取该连接。使用该连接,客户端能够与目的地通讯,往队列或话题发送/接收消息。让我们用一个例子来理解如何发送消息:

QueueConnectionFactory queueConnFactory = (QueueConnectionFactory) initialCtx.lookup ("primaryQCF");  Queue purchaseQueue = (Queue) initialCtx.lookup ("Purchase_Queue");  Queue returnQueue = (Queue) initialCtx.lookup ("Return_Queue");

② 目的地(Destination)

目的地指明消息被发送的目的地以及客户端接收消息的来源。JMS使用两种目的地,队列和话题。如下代码指定了一个队列和话题。

创建一个队列Session

QueueSession ses = con.createQueueSession (false, Session.AUTO_ACKNOWLEDGE); //get the Queue object   Queue t = (Queue) ctx.lookup ("myQueue"); //create QueueReceiver   QueueReceiver receiver = ses.createReceiver(t);

创建一个话题

TopicSession ses = con.createTopicSession (false, Session.AUTO_ACKNOWLEDGE); // get the Topic object   Topic t = (Topic) ctx.lookup ("myTopic"); //create TopicSubscriber   TopicSubscriber receiver = ses.createSubscriber(t);

(2)JMS连接对象(Connection)

Connection表示在客户端和JMS系统之间建立的连接(对TCP/IP socket的包装)。Connection可以产生一个或多个Session。跟ConnectionFactory一样,Connection也有两种类型:QueueConnection和TopicConnection。

连接对象封装了与JMS提供者之间的虚拟连接,如果我们有一个ConnectionFactory对象,可以使用它来创建一个连接。

Connection connection = connectionFactory.createConnection();

创建完连接后,需要在程序使用结束后关闭它:connection.close();

(3)JMS会话(Session)

Session 是我们对消息进行操作的接口,可以通过session创建生产者、消费者、消息等。Session 提供了事务的功能,如果需要使用session发送/接收多个消息时,可以将这些发送/接收动作放到一个事务中。

我们可以在连接创建完成之后创建session:

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

这里面提供了参数两个参数,第一个参数是是否支持事务,第二个是事务的类型。

(4)JMS消息生产者(Producer)

消息生产者由Session创建,用于往目的地发送消息。生产者实现MessageProducer接口,我们可以为目的地、队列或话题创建生产者。

MessageProducer producer = session.createProducer(dest);  MessageProducer producer = session.createProducer(queue);  MessageProducer producer = session.createProducer(topic);

创建完消息生产者后,可以使用send方法发送消息:

代码如下:

producer.send(message);

(5)JMS消息消费者(Consumer)

消息消费者由Session创建,用于接收被发送到Destination的消息。消费者实现MessageConsumer接口,我们可以为目的地、队列或话题创建消费者。

MessageConsumer consumer = session.createConsumer(dest);MessageConsumer consumer = session.createConsumer(queue);MessageConsumer consumer = session.createConsumer(topic);

(6)JMS消息监听器

JMS消息监听器是消息的默认事件处理者,他实现了MessageListener接口,该接口包含一个onMessage方法,在该方法中需要定义消息达到后的具体动作。通过调用setMessageListener方法,我们给指定的消费者定义了消息监听器。

Listener myListener = new Listener();consumer.setMessageListener(myListener);

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间:2020-03-24

浅谈java.util.concurrent包中的线程池和消息队列

1.java并发包介绍 JDK5.0(JDK1.5更名后)以后的版本引入高级并发特性,大多数的特性在java.util.concurrent包中,是专门用于多线程编程的,充分利用了现代多处理器和多核心系统的功能以编写大规模并发应用程序.主要包括原子量.并发集合.同步器.可重入,并对线程池的构造提供了强力的支持 2.线程池 java.util.concurrent.Executors提供了一个 java.util.concurrent.Executor接口的实现用于创建线程池 多线程技术主要解决

浅谈使用java实现阿里云消息队列简单封装

一.前言 最近公司有使用阿里消息队列的需求,为了更加方便使用,本人用了几天时间将消息队列封装成api调用方式以方便内部系统的调用,现在已经完成,特此记录其中过程和使用到的相关技术,与君共勉. 现在阿里云提供了两种消息服务:mns服务和ons服务,其中我认为mns是简化版的ons,而且mns的消息消费需要自定义轮询策略的,相比之下,ons的发布与订阅模式功能更加强大(比如相对于mns,ons提供了消息追踪.日志.监控等功能),其api使用起来更加方便,而且听闻阿里内部以后不再对mns进行新的开发

java实现消息队列的两种方式(小结)

实现消息队列的两种方式 Apache ActiveMQ官方实例发送消息 直接在Apache官网http://activemq.apache.org/download-archives.html下载ActiveMQ源码 下载解压后拿到java代码实例 然后倒入IDE 如下: 请认真阅读readme.md文件,大致意思就是把项目打成两个jar包,然后启动服务,然后同时运行打的两个jar包,然后就能看到具体的调用信息.打jar包时直接利用maven打就行了,不用修改代码. 启动服务: 利用Spring

Java中消息队列任务的平滑关闭详解

前言 消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构.目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ 消息队列应用场景 消息队列在实际应用中常用的使用场景:异步处理,应用解耦,流量削锋和消息通讯四个场景. 本文主要给大家介绍的是关于Java中消息队列任务平滑关闭的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细的介绍吧. 1.问题背

Java消息队列的简单实现代码

今天看到我们的招聘信息有对消息队列有要求,然后就思索了一翻,网上一搜一大堆. 我可以举个小例子先说明应用场景 假设你的服务器每分钟的处理量为200个,但客户端再峰值的时候可能一分钟会发1000个消息给你,这时候你就可以把他做成队列,然后按正常有序的处理,先进后出(LIFO),先进先出(FIFO)可根据自己的情况进行定夺 stack  先进后出(LIFO)——–Java 对应的类 Stack 队列 先进先出(FIFO)——–java对应的类Queue 这两种都可用Linkedl

Java利用Redis实现消息队列的示例代码

本文介绍了Java利用Redis实现消息队列的示例代码,分享给大家,具体如下: 应用场景 为什么要用redis? 二进制存储.java序列化传输.IO连接数高.连接频繁 一.序列化 这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: package Utils

浅谈Java消息队列总结篇(ActiveMQ、RabbitMQ、ZeroMQ、Kafka)

一.消息队列概述 消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构.目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ. 二.消息队列应用场景 以下介绍消息队列在实际应用中常用的使用场景.异步处理,应用解耦,流量削锋和消息通讯四个场景. 2.1异步处理 场景说明:用户注册后,需要发注册邮件和注册短信.传统的做法有两种 1.串行的方式;2.并行方式 a.串

java多线程消息队列的实现代码

本文介绍了java多线程消息队列的实现代码,分享给大家,希望对大家有帮助,顺便也自己留个笔记 1.定义一个队列缓存池: //static修饰的成员变量和成员方法独立于该类的任何对象.也就是说,它不依赖类特定的实例,被类的所有实例共享. private static List<Queue> queueCache = new LinkedList<Queue>(); 2.定义队列缓冲池最大消息数,如果达到该值,那么队列检入将等待检出低于该值时继续进行. private Integer

java多线程复制文件的实例代码

复制代码 代码如下: package com.test; import java.io.FileNotFoundException;  import java.io.IOException;  import java.io.RandomAccessFile; public class FileCoper {      private static final String _ORIGIN_FILE_MODE = "r"; private static final String _TAR

Java多线程下载文件实例详解

本文实例为大家分享了Java多线程下载文件的具体代码,供大家参考,具体内容如下 import java.io.File; import java.io.InputStream; import java.io.RandomAccessFile; import java.net.HttpURLConnection; import java.net.URL; public class MulThreadDownload { public static void main(String[] args)

PHP+memcache实现消息队列案例分享

memche消息队列的原理就是在key上做文章,用以做一个连续的数字加上前缀记录序列化以后消息或者日志.然后通过定时程序将内容落地到文件或者数据库. php实现消息队列的用处比如在做发送邮件时发送大量邮件很费时间的问题,那么可以采取队列.方便实现队列的轻量级队列服务器是:starling支持memcache协议的轻量级持久化服务器https://github.com/starling/starlingBeanstalkd轻量.高效,支持持久化,每秒可处理3000左右的队列http://kr.gi

Linux消息队列实现进程间通信实例详解

Java消息队列JMS实现原理解析

Linux消息队列实现进程间通信实例详解 一.什么是消息队列 消息队列提供了一种从一个进程向另一个进程发送一个数据块的方法.  每个数据块都被认为含有一个类型,接收进程可以独立地接收含有不同类型的数据结构.我们可以通过发送消息来避免命名管道的同步和阻塞问题.但是消息队列与命名管道一样,每个数据块都有一个最大长度的限制. Linux用宏MSGMAX和MSGMNB来限制一条消息的最大长度和一个队列的最大长度. 二.在Linux中使用消息队列 Linux提供了一系列消息队列的函数接口来让我们方便地使用

Java多线程编程实现socket通信示例代码

Java消息队列JMS实现原理解析

流传于网络上有关Java多线程通信的编程实例有很多,这一篇还算比较不错,代码可用.下面看看具体内容. TCP是Tranfer Control Protocol的 简称,是一种面向连接的保证可靠传输的协议.通过TCP协议传输,得到的是一个顺序的无差错的数据流.发送方和接收方的成对的两个socket之间必须建 立连接,以便在TCP协议的基础上进行通信,当一个socket(通常都是server socket)等待建立连接时,另一个socket可以要求进行连接,一旦这两个socket连接起来,它们就可以

消息队列 RabbitMQ 与 Spring 整合使用的实例代码

一.什么是 RabbitMQ RabbitMQ 是实现 AMQP(高级消息队列协议)的消息中间件的一种,最初起源于金融系统,用于在分布式系统中存储转发消息,在易用性.扩展性.高可用性等方面表现不俗.消息中间件主要用于组件之间的解耦,消息的发送者无需知道消息使用者的存在,反之亦然. RabbitMQ 是由 Erlang 语言开发,安装 RabbitMQ 服务需要先安装 Erlang 语言包. 二.如何与 Spring 集成 1. 我们都需要哪些 Jar 包? 抛开单独使用 Spring 的包不说,

原文 

https://www.zhangshengrong.com/p/l51gwGYl10/

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » Java消息队列JMS实现原理解析

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址