Java消息服务概要

jopen 9年前

一、消息中间件的作用

消息中间件，Message-Oriented Middleware，简称MOM。采用消息中间件的作用一般有两点：一是解耦，二是异步（起到削峰填谷的作用）

二、JMS的基本规范

1、消息传送模型

  • 点对点模型（基于队列的模式，如果有多个消费者，那么这些消费者轮流消费消息，达到负载均衡）

  • 发布订阅模型基于topic的发布/订阅模型较为流行，通常以此模型为主，外加点对点模型实现生产者/消费者的负载均衡。

2、消息接收模型

  • 推技术：消息服务器主动推送消息到消息消费者（JMS的Pub/Sub采用的是推技术）

  • 拉技术：消息消费者定时去消息服务器取消息（JMS接收者能够推送或拉取消息，取决于它是否使用异步onMessage回调或者是同步receive方法）

同步receive方式就是拉的方式，消费者主动去消息服务器取消息，异步的listener方式为推的方式。

同步receive的实例：

// Create the sender and send the message  QueueSender qSender = qSession.createSender(requestQ);  qSender.send(msg);  // Wait to see if the loan request was accepted or declined  String filter = "JMSCorrelationID = '" + msg.getJMSMessageID() + "'";  QueueReceiver qReceiver = qSession.createReceiver(responseQ, filter);  TextMessage tmsg = (TextMessage)qReceiver.receive(30000);  if (tmsg == null) {     System.out.println("Lender not responding");  } else {      System.out.println("Loan request was " + tmsg.getText());  }

3、基本概念

基本的概念：消息生产者、消息消费者、消息、目的地/topic/路由

（1）JMS1.X的api规范

（2）JMS2的api规范

4、消息格式

一条消息可分为 3 部分：消息头(Head)、属性(Attribute)和有效负载(Payload，即消息体)。

（1）消息头

这部分通常是结构化的数据，提供了和消息有关的元数据。

  • 消息唯一标识

  • 目的地

  • 传送方式（持久/非持久）

  • 消息接收时间

  • 消息失效时间

  • 消息优先级

  • 是否重发等

（2）消息属性

用于JMS的高级功能特性，以及存储一些应用特有的属性或者JMS厂商提供的额外功能属性。

  • A、高级功能：groupId/groupSeq，消息的分组聚合

  • B、应用自己添加的属性：比如username等

  • C、特殊厂商的额外功能属性

（3）消息体

一般可以有结构化的和非结构化的，JMS里头定义的比较详细，有Text、Object、Bytes、Stream、Map等格式。但是一般大多byte[]格式，自己可以选择自定义的序列化方式。

三、消息分发

1、消息选择器

在目的地上使用消息选择器，利用消息属性和消息头（无法使用消息体内的数据）作为条件表达式的准则，消息在目的地（队列/主题）分发给消费者之前就过滤好。（rabbitmq有基于routing key的表达式过滤方式，来选择接收哪几个topic的消息）

2、消费者控制还是生产者控制

（1）MessageFilter，消费者控制了消息过滤，并决定它要接受什么消息

  • 消息在目的地分发给消费者之前过滤。

  • 使用MessageFitler的优点是，具有更强的可伸缩性，假设增加了一个CustType为PLATINUM级别，那么只要增加一个相应的消息消费者就可以。

（2）Multiple Destination，多目的方式，生产者控制消息分发

  • 消息在发送给目的之前过滤，分发给不同目的地不同的消息

  • 如果是使用Multiple Destination方式，就是增加一个队列来保存PLATINUM，同时还得增加一个类来监听这个新的队列。

  • MultipleDestination的好处是生产者控制消息过滤分类，不容易出错，但是可扩展性稍差，MessageFilter是可扩展性好，但是容易出错。

3、未能过滤掉的消息如何处理

1）发布订阅模型：

这些消息不会传送给该订阅者，不论是持久订阅还是非持久订阅

2）点对点模型：

  • 未被消费者选择的所有消息，对该消费者都是不可见的。

  • 确认由发送者或发布者生产的所有消息各自都有和它们相关联的有效期，默认情况是永不过期，那么意味着如果一条消息被过滤掉，没有传送给消费者，它将在队列中永久驻留，可以通过设置生存时间来解决。

  • 有些厂商提出了死信队列（Dead Letter Queue，DLQ），或停用消息队列（Dead Message Queue，DMQ）的概念，来处理那些被认为无法传送的消息。

最简单的情况，消息传送系统将所有无法传送的消息放入DMQ，而应用程序负责监测它的内容。也可以支持管理型事件，能够在消息放入DMQ时通知应用程序。

对于rabbitmq，有一个dead letter的机制（当消息在一个队列中变成死信后，它能被重新publish到另一个Exchange，这个Exchange就是DLX）。

四、消息可靠性基础

1、消息保存转发机制

  • 保证传送机制（guaranteed delivery），确保即便发生了具备故障，预定消费者最终也会接收到这条消息。

  • 当消费者出故障时，将消息保存到持久化介质中，等待消费者恢复之后，从持久介质取出消息，转发给消费者

2、DeliveryMode

public interface DeliveryMode {      /** This is the lowest-overhead delivery mode because it does not require         * that the message be logged to stable storage. The level of JMS provider        * failure that causes a <CODE>NON_PERSISTENT</CODE> message to be lost is         * not defined.        *        * <P>A JMS provider must deliver a <CODE>NON_PERSISTENT</CODE> message         * with an         * at-most-once guarantee. This means that it may lose the message, but it         * must not deliver it twice.        */        static final int NON_PERSISTENT = 1;        /** This delivery mode instructs the JMS provider to log the message to stable         * storage as part of the client's send operation. Only a hard media         * failure should cause a <CODE>PERSISTENT</CODE> message to be lost.        */        static final int PERSISTENT = 2;  }

Message默认是PERSISTENT模式。

这个地方是JMS里头容易让人混乱的地方，其本质上，还是利用消息头的deliveryMode属性来标记的。

可以有两个地方可以设置：

（1）对于消息生产者来说

是发送消息的时候，设置是否持久化；

msg.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);

默认就是持久化模式的。

（2）对于消息消费者来说

是在接收消息之前设置传送模式。比如对于topic提供durableSubscriber的方法。

public interface TopicSession extends javax.jms.Session {      javax.jms.Topic createTopic(java.lang.String s) throws javax.jms.JMSException;        javax.jms.TopicSubscriber createSubscriber(javax.jms.Topic topic) throws javax.jms.JMSException;        javax.jms.TopicSubscriber createSubscriber(javax.jms.Topic topic, java.lang.String s, boolean b) throws javax.jms.JMSException;        javax.jms.TopicSubscriber createDurableSubscriber(javax.jms.Topic topic, java.lang.String s) throws javax.jms.JMSException;        javax.jms.TopicSubscriber createDurableSubscriber(javax.jms.Topic topic, java.lang.String s, java.lang.String s1, boolean b) throws javax.jms.JMSException;        javax.jms.TopicPublisher createPublisher(javax.jms.Topic topic) throws javax.jms.JMSException;        javax.jms.TemporaryTopic createTemporaryTopic() throws javax.jms.JMSException;        void unsubscribe(java.lang.String s) throws javax.jms.JMSException;  }

3、丢失重连

当客户端和服务器之间的网络连接丢失时，JMS提供者必须尽可能重新建立连接。如果该JMS提供者无法重新自动连接，在客户端调用某种能够引起网络流量的方法时，提供者必须抛出一个异常，向客户端通知这个情况。JMS提供了ExceptionListener接口，用于捕获丢失的连接（可以在捕获时重连），并向客户端通知这个情况。与MessageListener不同，MessageListener是与会话绑定在一起的。

五、消息确认机制

一般是自动确定、手动确定。

1、消息生产者发送消息时

TopicPublisher.publish和QueueSender.send方法是同步的，这些方法负责发送消息，同时进行阻塞，直到从消息服务器接收到一个确认为止。一旦接收到一个确认，执行线程就会恢复并返回方法，认为消息发送成功。底层确认对客户端编程模型来说是不可见的。如果在这个操作期间发生了一个故障情况，就会抛出一个异常，同时认为该消息未被传送（注意重新发送指的是消费服务器到消费者的重新发送）。

2、消息消费者接收消息时

如果会话是AUTO_ACKNOWLEDGE模式，当每个消费者获得消息时，JMS提供者的客户端运行时环境必须自动向服务器发送确认信息。如果服务器没有接收到这个确认信息，它就会认为该消息未被传送，并可能会试图重新传送。

3、消息服务器收发消息时

（1）服务器接收到生产者的消息，发送确认给生产者

确认消息从服务器发送到生产者，意味着服务器已经接收到该消息，并已经承担了传送它的责任。从JMS服务器的角度来看，发送到生产者的确认并未和消息传送直接关联。逻辑上，它们是两个独立的步骤。

  • A、对于持久消息来说，服务器将消息写入磁盘，然后在再通知生产者该消息已经被接收。

  • B、对于非持久消息，意味着服务器能够在接收到消息后立刻通知发送者，并将该消息存入内存。如果该消息的主题没有订阅者，根据厂商的不同，也可能会将该消息抛弃。

（2）消息消费者接收时确认

  • A、对于持久订阅者来说一直到消息服务器接收到所有的消息预定接收者的确认时，消息服务器才会认为该消息已经完成传送。要获得这些信息，必须对每个消费者都非常了解：哪一个客户端已经接收到每条消息，哪一个还没有接收到。一旦消息服务器将消息传送给所有的已知订阅者，并已分别从订阅者那里接收到确认，就会将这条消息从持久存储器中删除。如果持久订阅，而订阅者当前并未连接，那么消息服务器会将该消息保存起来，直到该订阅者变成可用状态或消息到期为止。甚至对于非持久消息来说，也是如此。也就是说，如果消息消费者设定为持久订阅，则不管消息生产者设定消息是否是持久的，当消费者不在时，总是保存，恢复时转发。

  • B、对于非持久性消息来说在消息服务器已经向发送者确认消息后，以及消息服务器有机会代表未连接的持久订阅者将消息写入磁盘之前，二者之间可能会有一个时间窗，如果JMS在这个时间窗内出现故障，该消息就可能丢失。

  • C、如果是使用持久消息时一个提供者可能会出现故障，但是会优雅恢复正常。由于消息保存在持久存储器中，它们并没有丢失，在提供者再次启动时，它们又会传送给消费者。如果是点对点队列，它们能够保证被传送出去，如果是发布订阅发送，只有消费者的订阅为持久性时，才能被保证传送出去，非持久订阅的传送行为因提供者不同而不同。

六、消息的事务机制

1、生产者提交消息的事务

  • 生产者在commit之前，JMS提供者不会开始向它的消费者传送消息，即使它已经从发送者那里接收到所有的消息。

  • 发送消息时，如果在发送消息的方法正常完成后没有调用commit方法，JMS提供者会从队列中删除这些消息，而这些消息并没有传送给消费者。

2、消费者消费消息的事务

  • 消息会尽可能快地传送给接收者，但是它们一直由JMS提供者保存，直到接收者在会话对象上发布commit为止，如果发生了故障或调用了rollback，提供会试图重新传送这些消息，这种情况下，这些消息会设置为重新传送标记。

  • 接收消息时，如果在接收消息方法正常完成后没有调用commit方法，消息就会被标记为未被传送，JMS提供者会将这些消息重新传送给消费者，并将JMSRedelivered标记为true，表示此前曾试图处理过这些消息。

</div>

来自： http://segmentfault.com/a/1190000004247661

相关经验