ActiveMQ学习笔记(一) JMS概要

2022-10-04 22:18:32

(一)什么是JMS 
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

  1. package javax.jms;
  2. public interface MessageProducer {
  3. void setDisableMessageID(boolean value) throws JMSException;
  4. boolean getDisableMessageID() throws JMSException;
  5. void setDisableMessageTimestamp(boolean value) throws JMSException;
  6. boolean getDisableMessageTimestamp() throws JMSException;
  7. void setDeliveryMode(int deliveryMode) throws JMSException;
  8. int getDeliveryMode() throws JMSException;
  9. void setPriority(int defaultPriority) throws JMSException;
  10. int getPriority() throws JMSException;
  11. void setTimeToLive(long timeToLive) throws JMSException;
  12. long getTimeToLive() throws JMSException;
  13. Destination getDestination() throws JMSException;
  14. void close() throws JMSException;
  15. void send(Message message) throws JMSException;
  16. void send(Message message, int deliveryMode, int priority, long timeToLive)
  17. throws JMSException;
  18. void send(Destination destination, Message message) throws JMSException;
  19. void send(Destination destination, Message message, int deliveryMode,
  20. int priority, long timeToLive) throws JMSException;
  21. }

MessageProducer不仅提供了发送消息的方法，同时也提供了设置消息头部的方法（JMSDeliveryMode、JMSPriority、JMSExpiration、TimeToLive）

3.2 Consumer

  1. package javax.jms;
  2. public interface MessageConsumer {
  3. String getMessageSelector() throws JMSException;
  4. MessageListener getMessageListener() throws JMSException;
  5. void setMessageListener(MessageListener listener) throws JMSException;
  6. Message receive() throws JMSException;
  7. Message receive(long timeout) throws JMSException;
  8. Message receiveNoWait() throws JMSException;
  9. void close() throws JMSException;
  10. }

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

  1. a value written as the row type can be read as the column type.
  2. |----------------------------------------------------------
  3. |        | boolean byte short int long float double String
  4. |----------------------------------------------------------
  5. |boolean |    X                                       X
  6. |byte    |          X     X    X   X                  X
  7. |short   |                X    X   X                  X
  8. |int     |                     X   X                  X
  9. |long    |                         X                  X
  10. |float   |                               X     X      X
  11. |double  |                                     X      X
  12. |String  |    X     X     X    X   X     X     X      X
  13. |----------------------------------------------------------

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 代码(忽略异常处理代码)

  1. import javax.jms.Connection;
  2. import javax.jms.ConnectionFactory;
  3. import javax.jms.Destination;
  4. import javax.jms.JMSException;
  5. import javax.jms.MessageProducer;
  6. import javax.jms.Session;
  7. import javax.jms.TextMessage;
  8. import org.apache.activemq.ActiveMQConnectionFactory;
  9. import org.apache.activemq.command.ActiveMQQueue;
  10. import org.apache.activemq.command.ActiveMQTextMessage;
  11. public class Main {
  12. public static void main(String[] args) throws JMSException {
  13. // 1
  14. ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
  15. // 2
  16. Connection connection = connectionFactory.createConnection();
  17. // 3
  18. connection.start();
  19. // 4
  20. Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  21. // 5
  22. Destination destination = new ActiveMQQueue("queue");
  23. // 6
  24. MessageProducer messageProducer = session.createProducer(destination);
  25. // 7
  26. TextMessage textMessage = new ActiveMQTextMessage();
  27. textMessage.setText("some text");
  28. messageProducer.send(textMessage);
  29. // 8
  30. messageProducer.close();
  31. session.close();
  32. connection.stop();
  33. connection.close();
  34. System.out.println("done!");
  35. }
  36. }
 
上一篇：mongobackup mongodb的增量备份恢复工具


下一篇：[NOI2009] 管道取珠