JMS是一种应用于异步消息传递的标准API

Posted 时间的朋友

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JMS是一种应用于异步消息传递的标准API相关的知识，希望对你有一定的参考价值。

JMS是一种应用于异步消息传递的标准API，作为Java平台的一部分，JMS可以允许不同应用、不同模块之间实现可靠、异步数据通信。

一些概念

JMS provider
    An implementation of the JMS interface for a Message Oriented Middleware (MOM). Providers are implemented as either a Java JMS implementation or an adapter to a non-Java MOM.
JMS client
    An application or process that produces and/or receives messages.
JMS producer/publisher
    A JMS client that creates and sends messages.
JMS consumer/subscriber
    A JMS client that receives messages.
JMS message
    An object that contains the data being transferred between JMS clients.
JMS queue
    A staging area that contains messages that have been sent and are waiting to be read. Note that, contrary to what the name queue suggests, messages don‘t have to be delivered in the order sent. A JMS queue only guarantees that each message is processed only once.
JMS topic
    A distribution mechanism for publishing messages that are delivered to multiple subscribers. 

在JMS中，支持两种消息模型，点对点（Point-to-point）和发布-订阅(Publish and subscribe)，这两种模式分别对应于JMS中的两种消息目标（Message Destination）:队列及主题。

在点对点模型中，每个消息都有一个发送者和一个接收者，消息中介（broker）收到发送者的消息，会将消息放入队列中，而接收者请求并接收队列中的一条消息后，这条消息就会从队列中删除。消息队列中的每条消息只能投递给一个接收者，但并不意味着只能使用一个接收者从队列中取消息，根据业务需要，可以使用多个接收者同时从队列中请求消息，分担处理压力。但是需要注意的是，单个接收者收到的消息是按照发送顺序的，多个接收者因为多线程的关系，并不能保证收到的消息一定是原序的。

在发布-订阅模式中，消息会发送给一个主题，但是与点对点模式不同的是消息不再只被投递给一个接收者，而是所有此主题的订阅者都会收到该消息。

JMS消息类型

在JMS1.1规范中，定义了五种消息类型，分别为：
1.StreamMessage ：消息体是 Java 流，写入和读出都是顺序的
2.MapMessage ：消息体包含 key-value 对， key 为 String ， value 为基本类型，可以通过迭代器访问
3.TextMessage ：消息体是 String
4.ObjectMessage ：消息体是可序列化的 Java 对象
5.BytesMessage ：消息体是字节数组
可以通过 message.clearBody() 来清除消息体；但在消费端，消息体是只读的，针对消息的写操作都会抛出 MessageNotWritableException 异常

JMS消息头
所有消息的消息头都具体相同的字段，用于 JMS Client 以及 JMS Provider 对它们进行区别以及进行消息路由
1.JMSDestination
消息发送的目的地（队列或主题）；创建消息时可以设置 JMSDestination ，但是在发送完成时其值会更新为发送方所指定的 JMSDestination ，也就是说发送前该字段会被忽略；当消息被消费时，该字段的值与在它被发送时被设置的值是相同的
以下所有示例均基于ActiveMQ
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建2个目的地
Destination destination = session.createQueue("JMS.DEMO");
Destination destination2 = session.createQueue("JMS.DEMO2");

// 创建生产者
MessageProducer publisher = session.createProducer(destination);

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);

// 创建消息
TextMessage message = session.createTextMessage("Test Message");
// 设置消息的目的地为destination2
message.setJMSDestination(destination2);

// 发送消息
publisher.send(message);

System.out.println(message.getJMSDestination());

代码中，通过 message.setJMSDestination(destination2); 设置了 message 的 JMSDestination 消息头属性值，我们再看看其输出结果
queue://JMS.DEMO

通过这个例子可以看出，虽然在发送前设置了消息的目的地，但是发送后消息的目的地被重置了
2.JMSDeliveryMode
指明消息的传输模式，有两种：
DeliveryMode.PERSISTENT ：保证消息仅传一次， JMS Provider 服务停止后消息不会丢失；
DeliveryMode.NON_PERSISTENT ：消息最多传一次，消息会因 JMS Provider 停止后丢失；
同 JMSDestination 一样，在发送前设置的会被忽略

看下面的例子
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建目的地
Destination destination = session.createQueue("JMS.DEMO");

// 创建生产者
MessageProducer publisher = session.createProducer(destination);

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);
// 发送PERSISTENT消息
publisher.send(session.createTextMessage("PERSISTENT MESSAGE"));

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 发送PERSISTENT消息
publisher.send(session.createTextMessage("NON_PERSISTENT MESSAGE"));

例子中分别发送了一条 PERSISTENT 的消息和一条 NON_PERSISTENT 的消息；当 Active MQ 重启后，启动消费端，收到的消息如下
PERSISTENT MESSAGE

该例子说明，在 JMS Provider 重启后， NON_PERSISTENT 消息丢失了，而 PERSISTENT 消息能正常被消费者消费
3.JMSMessageID
由 JMS Provider 指定的消息的唯一标识符；同上面的字段一样，在发送前设置的会被忽略，在发送完成时，由 JMS Provider 重置该字段

4.JMSReplyTo
发送端在发送消息时，可以指定该属性（为一个 JMSDestination ），表示期望收到客户端的响应；是否响应由消费端决定
如下面的例子：
发送端： 
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建目的地
Destination destination = session.createQueue("JMS.DEMO");
Destination destination2 = session.createQueue("JMS.DEMO3");

// 创建生产者
MessageProducer publisher = session.createProducer(destination);

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);

// 创建消息
TextMessage message = session.createTextMessage("Test Message");
message.setJMSReplyTo(destination2);
// 发送消息
publisher.send(message);

接收端（可以根据情况决定是否需要回复）
public void onMessage(Message message) {
    try {
        System.out.println("Receive message: " + message);
        if (message.getJMSReplyTo() != null) {
            session.createProducer(message.getJMSReplyTo()).send(session.createTextMessage("This is a reply to"
                                                                                           + message.getJMSReplyTo()));
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

5.JMSRedelivered
当消费者收到带有 JMSRedelivered 的消息头时，表明该消息在过去传输过但没有被确认
JMS Provider 必须对该字段进行设置，当为 true 时即告知消费者该消息是重传的，消费者需要自行处理重复的消息

6.JMSExpiration
消息的过期时间，其值为当前时间加上存活时间（毫秒）；当存活时间设置为 0 时，该字段的值也被设置为 0 ，表示永不过期；
消费端在一般情况下都不会接收到过期的消息，但 JMS Provider 并不保证这一点；
下面的例子说明了如何设置消息的过期时间
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建目的地
Destination destination = session.createQueue("JMS.DEMO");

// 创建生产者
MessageProducer publisher = session.createProducer(destination);

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);

// 创建消息
TextMessage message = session.createTextMessage("Test Message");
// 发送消息
publisher.setTimeToLive(5000);
publisher.send(message);

7.JMSPriority
消息的优先级， 0 代表最低优先级， 9 代表最高优先级；一般 0~4 为普通优先级， 5~9 为加快优先级
JMS 规范里并没有要求 JMS Provider 严格按这个优先级来实现，但是尽可能实现加快优先级消息的传输在普通消息的前面
同 JMSDestination 一样，该字段在发送前被忽略，在发送完成时重置


消息属性

除了前面提到的消息头以外， JMS 消息还提供了对“属性值对”的支持，以对消息头进行扩展；消息属性主要用于消息选择器 (message selector 详见下文 ) 

1.属性名
属性名必须服务消息选择器的命名规则

2.属性值
可以是基本类型及其对象类型以及 Map 、 List 和 String

下面的例子中，消息带 HashMap 的属性
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创建目的地
Destination destination = session.createQueue("JMS.DEMO");

// 创建生产者
MessageProducer publisher = session.createProducer(destination);

// 设置传输模式
publisher.setDeliveryMode(DeliveryMode.PERSISTENT);

// 创建消息
TextMessage message = session.createTextMessage("Test Message");
// 发送消息
message.setObjectProperty("myProp", new HashMap() {

    {
        this.put("key1", "value1");
        this.put("key2", "value2");
    }
});
publisher.send(message);

3.清除属性
JMS 不能清除单个属性，但可以通过 Message.clearProperties() 方法清除所有消息属性

JMS实现（Provider implementations）

要使用JMS，必须要有相应的实现来管理session以及队列，从Java EE1.4开始，所有的Java EE应用服务器必须包含一个JMS实现。
以下是一些JMS实现：
    Apache ActiveMQ
    Apache Qpid, using AMQP
    BEA Weblogic (part of the Fusion Middleware suite) and Oracle AQ from Oracle
    EMS from TIBCO
    FFMQ, GNU LGPL licensed
    JBoss Messaging and HornetQ from JBoss
    JORAM, from the OW2 Consortium
    Open Message Queue, from Sun Microsystems
    OpenJMS, from The OpenJMS Group
    RabbitMQ, using AMQP
    Solace JMS from Solace Systems
    SonicMQ from Progress Software
    StormMQ, using AMQP
    SwiftMQ
    Tervela
    Ultra Messaging from 29 West (acquired by Informatica)
    webMethods from Software AG
    WebSphere Application Server from IBM, which provides an inbuilt default messaging provider known as the Service Integration Bus (SIBus), or which can connect to WebSphere MQ as a JMS provider [5]
    WebSphere MQ (formerly MQSeries) from IBM

无论是持久消息，还是非持久消息，如果消息没有对应的消费者，那么activeMQ会认为这些消息无用，直接删除。
持久订阅者/非持久订阅者，只影响离线的时候消息(包括持久消息和非持久消息)是否能接收到，和消息是否持久无关；持久消息/非持久消息，只是影响jms provider宕机后。消息是否会丢失，如果永远不会宕机，那么持久消息和非持久消息没有区别。




以上是关于JMS是一种应用于异步消息传递的标准API的主要内容，如果未能解决你的问题，请参考以下文章

消息队列-OPEN MQ

JMS入门简介

4.java JMS技术

Java消息服务-JMS

一：ActiveMQ知识整理

消息中间件 — JMS规范