jms即Java消息服务(Java Message Service)应用程序接口是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持。
(二)常见的JMS提供商有哪些?
- IBM 的 MQSeries
- BEA 的 Weblogic JMS service
- Apache 的 ActiveMQ
本系列都采用ActiveMQ
(三)核心API
JMS第一个版本是1998年制定的,最后一个版本是在2002年制定的JMS1.1。
本文基于JMS1.1
- JMS Provider 完全使用Java对JMS接口的一个实现,上面所说的三个就是
- JMS Client 一个完全使用JMS编写的发送和接收消息的客户端(必然用Java实现)
- Non-JMS Client 使用JMS提供商提供的本地客户端API编写收发消息,而不是JMS提供的API
- JMS Producer 一个创建和发送JMS消息的客户端应用
- JMS Consumer 一个接收并处理JMS消息的客户端应用
- JMS Message JMS基础概念,JMS客户端发送和接收的消息
- JMS Domains 两种类型的消息,分别为点对点(point-to-point)和发布/订阅(publish/subscribe)
- Administered Object 对JMS对象进行配置
3.1 Provider
- package javax.jms;
- public interface MessageProducer {
- void setDisableMessageID(boolean value) throws JMSException;
- boolean getDisableMessageID() throws JMSException;
- void setDisableMessageTimestamp(boolean value) throws JMSException;
- boolean getDisableMessageTimestamp() throws JMSException;
- void setDeliveryMode(int deliveryMode) throws JMSException;
- int getDeliveryMode() throws JMSException;
- void setPriority(int defaultPriority) throws JMSException;
- int getPriority() throws JMSException;
- void setTimeToLive(long timeToLive) throws JMSException;
- long getTimeToLive() throws JMSException;
- Destination getDestination() throws JMSException;
- void close() throws JMSException;
- void send(Message message) throws JMSException;
- void send(Message message, int deliveryMode, int priority, long timeToLive)
- throws JMSException;
- void send(Destination destination, Message message) throws JMSException;
- void send(Destination destination, Message message, int deliveryMode,
- int priority, long timeToLive) throws JMSException;
- }
MessageProducer不仅提供了发送消息的方法,同时也提供了设置消息头部的方法(JMSDeliveryMode、JMSPriority、JMSExpiration、TimeToLive)
3.2 Consumer
- package javax.jms;
- public interface MessageConsumer {
- String getMessageSelector() throws JMSException;
- MessageListener getMessageListener() throws JMSException;
- void setMessageListener(MessageListener listener) throws JMSException;
- Message receive() throws JMSException;
- Message receive(long timeout) throws JMSException;
- Message receiveNoWait() throws JMSException;
- void close() throws JMSException;
- }
JMS客户端使用JMS MessageConsumer类来接收和处理消息。可以使用receive方法处理同步消息,也可以采用MessageListener的实现来处理异步消息。
MessageConsumer没有设置目标(destination)的方法,在消费者被创建的时候(session.createConsumer())需要提供默认的目标。
3.3 Message
JMS允许消息承载多种形式的信息,可以是文本形式,也可以是二进制的数据或者是在头部提供特殊的属性设置。
JMS由以下三部分组成。
- header消息头
- properties属性对
- payload实际负载
3.3.1 Header 可以通过API设置
- JMSDestination 消息投递目标
- JMSDeliveryMode 可以设置消息投递模式(persistent/non-persistent)。建议使用持久化方式,以获得更佳的可靠性。
- JMSExpiration 消息过期的时间。可以使用MessageProducer.setTimeToLive()设置默认的过期时间,也可以使用MessageProducer.send()方法设置个别消息的过期时间。0表示消息永不过期。
- JMSMessageID 一个能够唯一标识一条消息的字符串,这个字符串被JMS提供者签发,不建议应用依赖这个ID编程。
- JMSPriority 0到9,9为最高。JMS提供商只是尽量地将优先级高的消息先投递。不建议应用依赖这个次序编程。
- JMSTimestamp 这个头选项标志了消息提供者向JMS提供者发送消息的时间。MessageProducer.setDisableMessageTimestamp()可取消。
3.3.2 Properties 附加的头部信息。JMS提供使用Java原始模型设置属性的方法包括boolean、byte、short、int、long、float、double和String Object
- a value written as the row type can be read as the column type.
- |----------------------------------------------------------
- | | boolean byte short int long float double String
- |----------------------------------------------------------
- |boolean | X X
- |byte | X X X X X
- |short | X X X X
- |int | X X X
- |long | X X
- |float | X X X
- |double | X X
- |String | X X X X X X X X
- |----------------------------------------------------------
3.3.3 Payload 消息体。JMS一共定义了5种。
- Stream 流
- Map 名值对 key为String,value为序列化对
- Text 文本
- Object 可序列化对象
- Bytes 二进制序列
3.3.4 消息选择器
很多情况下,JMS客户端订阅了一个目标(Destination)的消息,但是客户端可能仅想消费其中一部分,此时需要对消息进行过滤。可以使用消息的头部或者属性字段对消息进行过滤。JMS客户端工具包提供的选择器(selectors),会通知JMS 提供商(Provider),消费者只想接收包含了特定属性字段的消息。消息选择器通过对消息头中的属性进行过滤,允许JMS客户端指定哪些消息是其关注的消息。选择器使用SQL92表达式的子集定义。消息选择器使用布尔逻辑,通过对消息的头部属性和属性字段为评判目标进行简单布尔计算。不满足表达式的消息将不被投递到客户端。消息选择器不能对消息的实际负载进行过滤,只能对消息头部的属性和属性字段进行过滤。
3.3.5 JMS Domains
JMS有两种类型的消息,分别是发布/订阅式和点对点式。
3.3.5.1 Queue
点对点式的消息传输中目标(Destination)将被认为是一个队列,消息可以同步的或者异步的被发送和接收。队列中的每条消息将仅会投递给一个消费者,并且仅投递一次(once-and-only-once 语义)。这很像发送email一样。消费者可以使用MessageConsumer.receive()同步接收队列中的消息,也可以实现MessageListener的MessageConsumer.setMessageListener()方法异步接收消息。队列将会存储所有投递来的消息,直到消息被投递出去或者消息已经过期。订阅消息可以通过MessageConsumer.receive()同步的从主题(topic)处接收消息,也可以通过实现MessageListener的setMessageListener方法异步接收订阅主题。多个消费者可以注册同一个消息队列接收消息,但是一条PTP消息仅会发给一个消费者。上图中需要注意,一个PTP消息的发送者发送的消息仅会被一个消费者接收,并不是所有注册到这个队列的消费者都会接收到同一消息。JMS提供者保证消息有且仅有一次被投递到注册的消费者。JMS提供商会在众多消费者之间做到一种负载平衡。
3.3.5.2 Topic
发布/订阅使用远程的主题(topic)目的地。发布者将消息发布给远程的主题,订阅者注册接收相应主题的消息。所有的发向主题的消息都会被以一种推的方式(push model)投递给所有的接收者。订阅消息可以通过MessageConsumer.receive()同步的从主题(topic)处接收消息,也可以通过实现MessageListener的setMessageListener方法异步接收订阅主题。如果不显示指明保持(durability)消息,主题消息将不会被保持。可以使用具有生命期(durable)的订阅。使用具有生命期的订阅时,如果有订阅者在消息发布时没有与JMS提供者连接,JMS将会为这种订阅者存储消息,当订阅者再次连接到JMS提供者时,这个订阅者将会收到失去连接期间的所有没过期的消息。生命期订阅允许发送消息时,接收者的非连接状态。
3.3.5.3 生命期消息(Message Durability)
生命期消息仅适用于发布/订阅消息域,当客户连接到主题时,他们可以决定是使用有生命期的(durable)还是无生命期的(non-durable)订阅。
- Durable Subscription 当接收有生命期主题订阅的消费者在消息发布时处于不可用状态,JMS提供者将会对订阅的主题进行存储,当消费者状态可用时,会接收到之前订阅的消息。
- Non-Durable Subscription 在消费者不可用期间发布的所有消息,都将丢失。
注意:消息持久化(persistence)和生命期消息(Message Durability)概念不同。持久化是一个独立任何域的概念。主要用于JMS提供者崩溃时消息的恢复。可以通过JMSDeliveryMode属性设置消息的持久化和非持久化属性。
3.4 管理员对象
JMS规范定义了两种管理员对象:ConnectionFactory和Destination。
- ConnectionFactory:SPI的*工厂,根据JMS提供商的不同,这个对象得到方式会有不同。
- Destination Destination:对象封装了提供者相关的消息发送和接收的地址。尽管Destination是Session创建的,但是它们的生命期与创建Session的连接相匹配(就是说Session终结了,连接没有终结,Destination仍然有效)。
(四) Native JMS的使用
4.1 步骤(以发送为例)
- 获得JMS连接工厂(connection Factory)
- 使用JMS工厂创建JMS连接
- 启动JMS连接
- 使用JMS连接创建JMS 会话(Session)
- 获得一个JMS的目的地
- 创建JMS 发送者(Producer)并指明发送目标
- 发送JMS消息
- 关闭JMS资源(connection、Session、producer、Consumer)
4.2 代码(忽略异常处理代码)
- import javax.jms.Connection;
- import javax.jms.ConnectionFactory;
- import javax.jms.Destination;
- import javax.jms.JMSException;
- import javax.jms.MessageProducer;
- import javax.jms.Session;
- import javax.jms.TextMessage;
- import org.apache.activemq.ActiveMQConnectionFactory;
- import org.apache.activemq.command.ActiveMQQueue;
- import org.apache.activemq.command.ActiveMQTextMessage;
- public class Main {
- public static void main(String[] args) throws JMSException {
- // 1
- ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
- // 2
- Connection connection = connectionFactory.createConnection();
- // 3
- connection.start();
- // 4
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- // 5
- Destination destination = new ActiveMQQueue("queue");
- // 6
- MessageProducer messageProducer = session.createProducer(destination);
- // 7
- TextMessage textMessage = new ActiveMQTextMessage();
- textMessage.setText("some text");
- messageProducer.send(textMessage);
- // 8
- messageProducer.close();
- session.close();
- connection.stop();
- connection.close();
- System.out.println("done!");
- }
- }