ActiveMQ

编程技术  /  houtizong 发布于 3年前   122

一、JMS
1. JMS简介
  JMS即Java消息服务（Java Message Service），是一个是Java平台上有关面向消息中间件(MOM)的技术规范，它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发，用于在两个应用程序之间或分布式系统中发送消息，进行异步通信。

2. JMS对象模型要素：
  1）连接工厂：连接工厂（ConnectionFactory）是由管理员创建，并绑定到JNDI树中。客户端使用JNDI查找连接工厂，然后利用连接工厂创建一个JMS连接。例如ActiveMQ提供的
ActiveMQConnectionFactory。
  2）JMS连接：JMS连接（Connection）表示JMS客户端和服务器端之间的一个活动的连接，是由客户端通过调用连接工厂的方法建立的。
  3）JMS会话：JMS Session是生产和消费消息的一个单线程上下文。会话用于创建消息生产者（producer）、消息消费者（consumer）和消息（message）等。会话提供了一个事务性的上下文，在这个上下文中，一组发送和接收被组合到了一个原子操作中。
  4）JMS目的地：JMS目的地（Destination），又称为消息队列，是实际的消息源。
  5）JMS生产者和消费者：生产者（Message Producer）和消费者（Message Consumer）对象由Session对象创建，用于发送和接收消息。
  6）JMS消息：JMS消息由以下三部分组成：
      • 消息头：每个消息头字段都有相应的getter和setter方法。
      • 消息属性：如果需要除消息头字段以外的值，那么可以使用消息属性。
      • 消息体：JMS定义的消息类型有TextMessage、MapMessage、BytesMessage、StreamMessage和ObjectMessage。
  7）JMS消息类型：
      点对点（Point-to-Point）：在点对点的消息系统中，消息分发给一个单独的使用者。点对点消息往往与队列（javax.jms.Queue）相关联。
      发布/订阅（Publish/Subscribe）：发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递。生产者发布事件，而使用者订阅感兴趣的事件，并使用事件。该类型消息一般与特定的主题（javax.jms.Topic）关联。

3. JMS的可靠性机制
  1）确认
    JMS消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。 在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：
      Session.AUTO_ACKNOWLEDGE：当客户成功的从receive方法返回的时候，或者从MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息。
      Session.CLIENT_ACKNOWLEDGE： 客户通过消息的acknowledge方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行：确认一个被消费的消息将自动确认所有已被会话消 费的消息。例如，如果一个消息消费者消费了10个消息，然后确认第5个消息，那么所有10个消息都被确认。
      Session.DUPS_ACKNOWLEDGE： 该选择只是会话迟钝第确认消息的提交。如果JMS provider失败，那么可能会导致一些重复的消息。如果是重复的消息，那么JMS provider必须把消息头的JMSRedelivered字段设置为true。
  2）持久性
    JMS 支持以下两种消息提交模式：
      PERSISTENT：指示JMS provider持久保存消息，以保证消息不会因为JMS provider的失败而丢失。
      NON_PERSISTENT：不要求JMS provider持久保存消息。
  3）优先级
    可以使用消息优先级来指示JMS provider首先提交紧急的消息。优先级分10个级别，从0（最低）到9（最高）。如果不指定优先级，默认级别是4。需要注意的是，JMS provider并不一定保证按照优先级的顺序提交消息。
  4）消息过期
    可以设置消息在一定时间后过期，默认是永不过期。
  5）临时目的地
    可以通过会话上的createTemporaryQueue方法和createTemporaryTopic方法来创建临时目的地。它们的存在时间只限于创建它们的连接所保持的时间。只有创建该临时目的地的连接上的消息消费者才能够从临时目的地中提取消息。
  6）持久订阅
    首先消息生产者必须使用PERSISTENT提交消息。客户可以通过会话上的createDurableSubscriber方法来创建一个持久订阅，该方法的第一个参数必须是一个topic，第二个参数是订阅的名称。 JMS provider会存储发布到持久订阅对应的topic上的消息。如果最初创建持久订阅的客户或者任何其它客户使用相同的连接工厂和连接的客户ID、相同的主题和相同的订阅名再次调用会话上的createDurableSubscriber方法，那么该持久订阅就会被激活。JMS provider会象客户发送客户处于非激活状态时所发布的消息。 持久订阅在某个时刻只能有一个激活的订阅者。持久订阅在创建之后会一直保留，直到应用程序调用会话上的unsubscribe方法。
  7）本地事务
    在一个JMS客户端，可以使用本地事务来组合消息的发送和接收。JMS Session接口提供了commit和rollback方法。事务提交意味着生产的所有消息被发送，消费的所有消息被确认；事务回滚意味着生产的所有消息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。 事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。 需要注意的是，如果使用请求/回复机制，即发送一个消息，同时希望在同一个事务中等待接收该消息的回复，那么程序将被挂起，因为知道事务提交，发送操作才会真正执行。 需要注意的还有一个，消息的生产和消费不能包含在同一个事务中。

二、ActiveMQ
1. ActiveMQ简介
  在设计分布式应用程序时，应用程序间的耦合（或称集成）方式很重要。耦合意味着两个或者多个应用程序或系统的相互依赖关系。一种简单的方式是在所有的应用程序中从架构上设计他们与其他应用程序间的交叉实现。这样必然导致，一个应用程序的改变，直接导致另一个应用程序的改变。按照这种方式集成的应用是一种紧耦合的应用。一个应用的改变不会影响到其他应用的集成方式被称为是松耦合的集成方式。简单的说，松耦合应用程序集成能够更容易的处理不可预见的应用变化。
  ActiveMQ是Apache下一个开源的，实现了JMS1.1（Java消息服务）规范的，面向消息（MOM）的中间件，为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。它的设计目标是提供标准的，面向消息的，能够跨越多语言和多系统的应用集成消息通信中间件。ActiveMQ实现了JMS标准并提供了很多附加的特性。这些附加的特性主要包括：
    JMX管理（java Management Extensions，即java管理扩展）
    主从管理（master/slave，这是集群模式的一种，主要体现在可靠性方面，当主中介（代理）出现故障，那么从代理会替代主代理的位置，不至于使消息系统瘫痪）
    消息组通信（同一组的消息，仅会提交给一个客户进行处理）
    有序消息管理（确保消息能够按照发送的次序被接受者接收）
    消息优先级（优先级高的消息先被投递和处理）
    订阅消息的延迟接收（订阅消息在发布时，如果订阅者没有开启连接，那么当订阅者开启连接时，消息中介将会向其提交之前的，其未处理的消息）
    接收者处理过慢（可以使用动态负载平衡，将多数消息提交到处理快的接收者，这主要是对PTP消息所说）
    虚拟接收者（降低与中介的连接数目）
    成熟的消息持久化技术（部分消息需要持久化到数据库或文件系统中，当中介崩溃时，信息不会丢失）
    支持游标操作（可以处理大消息）
    支持消息的转换
    通过使用Apache的Camel可以支持EIP
    使用镜像队列的形式轻松的对消息队列进行监控

2. Broker
  Broker是一种分布式系统的体系结构模式，它能够对激发远程对象服务的组件进行解耦，封装远程对象之间的通信，而且能够提供一种在异质的系统之间相对独立的分布式计算环境。Broker是消息的传送者，它把客户端的请求发送到服务端，把服务端的返回的结果和异常发给客户端。这要求服务端的服务应该能够在Broker进行注册，这样Broker才能够进行服务的目录查找。
  Running Broker：ActiveMQ发布包中bin目录中包含一个名为activemq的脚本，直接运行这个脚本就可以启动一个broker。
  Embedded Broker：可以通过在应用程序中以编码的方式启动broker，例如：
    BrokerService broker = new BrokerService();
    broker.setName("fred"); //有多个broker时，需要为broker设置一个名字
    broker.addConnector("tcp://localhost:61616"); 
    broker.start();
    或者通过BrokerFactory来创建broker：
    BrokerService broker = BrokerFactory.createBroker(new URI(someURI));
  Monitoring Broker：包括JMX监控broker、Web Console、Advisory Message标准的JMS消息监控系统、Command Agent。

3. Transport（传输）
  ActiveMQ目前支持的transport有：
  1）VM transport：它允许在VM内部通信，从而避免了网络传输的开销。这时候采用的连接不是socket连接，而是直接地方法调用。 第一个创建VM 连接的客户会启动一个embed VM broker，接下来所有使用相同的broker name的VM连接都会使用这个broker。当这个broker上所有的连接都关闭的时候，这个broker也会自动关闭。
  2）TCP transport：它允许客户端通过TCP socket连接到远程的broker。
  3）Failover Transport：它是一种重新连接的机制，它工作于其它transport的上层，用于建立可靠的传输。它的配置语法允许制定任意多个复合的URI。 Failover transport会自动选择其中的一个URI来尝试建立连接。如果没有成功，那么会选择一个其它的URI来建立一个新的连接。
  4）Discovery transport：它是可靠的tranport。它使用Discovery transport来定位用来连接的URI列表。

4. Persistence（持久性）
  Persistence包括：
  1）AMQ Message Store：它是ActiveMQ缺省的持久化存储。Message commands被保存到transactional journal（由rolling data logs组成）。Messages被保存到data logs中，同时被reference store进行索引以提高存取速度。Date logs由一些单独的data log文件组成，缺省的文件大小是32M，如果某个消息的大小超过了data log文件的大小，那么可以修改配置以增加data log文件的大小。如果某个data log文件中所有的消息都被成功消费了，那么这个data log文件将会被标记，以便在下一轮的清理中被删除或者归档。
  2）Kaha Persistence：它是一个专门针对消息持久化的解决方案。它对典型的消息使用模式进行了优化。在Kaha中，数据被追加到data logs中。当不再需要log文件中的数据的时候，log文件会被丢弃。
  3）JDBC Persistence：目前支持的数据库有Apache Derby, Axion, DB2,HSQL, Informix, MaxDB, MySQL, Oracle, Postgresql, SQLServer, Sybase。 如果你使用的数据库不被支持，那么可以调整StatementProvider 来保证使用正确的SQL方言（flavour of SQL）。
  4）Disable Persistence：禁用持久化。

5. Security
  ActiveMQ支持可插拔的安全机制，用以在不同的provider之间切换。
  1）Simple Authentication Plugin：它适用于简单的认证需求，或者用于建立测试环境。它允许在XML配置文件中指定用户、用户组和密码等信息。
  2）JAAS Authentication Plugin：它依赖标准的JAAS机制来实现认证。通常情况下，你需要通过设置java.security.auth.login.config系统属性来 配置login modules的配置文件。如果没有指定这个系统属性，那么JAAS Authentication Plugin会缺省使用login.config作为文件名。
  3）Custom Authentication Implementation：它可以通过编码的方式为ActiveMQ增加认证功能。例如编写一个类继承自XBeanBrokerService。
  4）Authorization Plugin：可以通过Authorization Plugin为认证后的用户授权。

6. Clustering（集群）
  ActiveMQ对集群的支持包括：
  1）Queue consumer clusters： ActiveMQ支持订阅同一个queue的consumers上的集群。如果一个consumer失效，那么所有未被确认 （unacknowledged）的消息都会被发送到这个queue上其它的consumers。如果某个consumer的处理速度比其它 consumers更快，那么这个consumer就会消费更多的消息。
  2）Broker clusters：一个常见的场景是有多个JMS broker，有一个客户连接到其中一个broker。如果这个broker失效，那么客户会自动重新连接到其它的broker。在ActiveMQ中使用failover://协议来实现这个功能。ActiveMQ3.x版本的reliable://协议已经变更为failover://。
  3）Master Slave：在一个网络内运行多个brokers或者stand alone brokers时存在一个问题，这就是消息在物理上只被一个broker持有，因此当某个broker失效，那么你只能等待直到它重启后，这个 broker上的消息才能够被继续发送（如果没有设置持久化，那么在这种情况下，消息将会丢失）。Master Slave 背后的想法是，消息被复制到slave broker，因此即使master broker遇到了像硬件故障之类的错误，你也可以立即切换到slave broker而不丢失任何消息。
  1）Pure Master Slave：
    Slave broker消费master broker上所有的消息状态，例如消息、确认和事务状态等。只要slave broker连接到了master broker，它不会（也不被允许）启动任何network connectors或者transport connectors，所以唯一的目的就是复制master broker的状态。
    Master broker只有在消息成功被复制到slave broker之后才会响应客户。例如，客户的commit请求只有在master broker和slave broker都处理完毕commit请求之后才会结束。
    当master broker失效的时候，slave broker有两种选择，一种是slave broker启动所有的network connectors和transport connectors，这允许客户端切换到slave broker；另外一种是slave broker停止。这种情况下，slave broker只是复制了master broker的状态。
    客户应该使用failover transport并且应该首先尝试连接master broker。
    只能有一个slave broker连接到master broker。
    在因master broker失效而导致slave broker成为master之后，之前的master broker只有在当前的master broker（原slave broker）停止后才能重新生效。
    Master broker失效后而切换到slave broker后，最安全的恢复master broker的方式是人工处理。首先要停止slave broker（这意味着所有的客户也要停止）。然后把slave broker的数据目录中所有的数据拷贝到master broker的数据目录中。然后重启master broker和slave broker。
  2）Shared File System Master Slave：
    如果你使用SAN或者共享文件系统，那么你可以使用Shared File System Master Slave。基本上，你可以运行多个broker，这些broker共享数据目录。当第一个broker得到文件上的排他锁之后，其它的broker便会 在循环中等待获得这把锁。客户端使用failover transport来连接到可用的broker。当master broker失效的时候会释放这把锁，这时候其中一个slave broker会得到这把锁从而成为master broker。
  3）JDBC Master Slave：
    JDBC Master Slave的工作原理跟Shared File System Master Slave类似，只是采用了数据库作为持久化存储。

7. Features（特性）
  ActiveMQ包含了很多功能强大的特性，下面简要介绍其中的几个：
  1）Exclusive Consumer：Broker会从多个consumers中挑选一个consumer来处理queue中所有的消息，从而保证了消息的有序处理。如果这个consumer 失效，那么broker会自动切换到其它的consumer。
  2）Message Groups：逻辑上，Message Groups 可以看成是一种并发的Exclusive Consumer。跟所有的消息都由唯一的consumer处理不同，JMS 消息属性JMSXGroupID 被用来区分message group。Message Groups特性保证所有具有相同JMSXGroupID 的消息会被分发到相同的consumer（只要这个consumer保持active）。另外一方面，Message Groups特性也是一种负载均衡的机制。
  3）JMS Selectors：它用于在订阅中，基于消息属性对进行消息的过滤，由SQL92语法定义。
  4）Pending Message Limit Strategy：慢消费者会在非持久的topics上导致问题：一旦消息积压起来，会导致broker把大量消息保存在内存中，broker也会因此而变慢。未来 ActiveMQ可能会实现磁盘缓存，但是这也还是会存在性能问题。目前ActiveMQ使用Pending Message Limit Strategy来解决这个问题。除了prefetch buffer之外，你还要配置缓存消息的上限，超过这个上限后，新消息到来时会丢弃旧消息。通过在配置文件的destination map中配置PendingMessageLimitStrategy，可以为不用的topic namespace配置不同的策略。
  5）Composite Destinations：它允许用一个虚拟的destination 代表多个destinations。例如你可以通过composite destinations在一个操作中同时向12个queue发送消息。在composite destinations中，多个destination之间采用","分割。
  6）Mirrored Queues：ActiveMQ支持Mirrored Queues。Broker会把发送到某个queue的所有消息转发到一个名称类似的topic，因此监控程序可以订阅这个mirrored queue topic。为了启用Mirrored Queues，首先要将BrokerService的useMirroredQueues属性设置成true，然后可以通过 destinationInterceptors设置其它属性，如mirror topic的前缀，缺省是"VirtualTopic.Mirror."。
  7）Wildcards：Wildcards用来支持联合的名字分层体系（federated name hierarchies）。它不是JMS规范的一部分，而是ActiveMQ的扩展。ActiveMQ支持以下三种wildcards：
    "." 用于作为路径上名字间的分隔符。
    "*" 用于匹配路径上的任何名字。
    ">" 用于递归地匹配任何以这个名字开始的destination。

三、ActiveMQ的简单示例
1.下载ActiveMQ
  官网下载地址：http://activemq.apache.org/

2.运行ActiveMQ
  解压下载文件 apache-activemq-5.5.1-bin.zip，然后运行 apache-activemq-5.5.1\bin\activemq.bat 文件启动ActiveMQ，然后登陆：http://localhost:8161/admin/，创建一个Queue，此处命名 JiangQueue。

3. 创建项目
  创建一个Java项目，并引入apache-activemq-5.5.1\lib 下的 activemq-core-5.5.1.jar，geronimo-j2ee-management_1.1_spec-1.0.1.jar，geronimo-jms_1.1_spec-1.1.1.jar，log4j-1.2.14.jar，slf4j-api-1.5.11.jar，slf4j-log4j12-1.5.11.jar。

4. 发送端测试

package com.jiang.activemq;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.DeliveryMode;import javax.jms.Destination;import javax.jms.MessageProducer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;public class Sender {private static final int SEND_NUMBER = 5;public static void main(String[] args) {// ConnectionFactory ：连接工厂，JMS 用它创建连接ConnectionFactory connectionFactory;// Connection ：JMS 客户端到JMS Provider 的连接Connection connection = null;// Session： 一个发送或接收消息的线程Session session;// Destination ：消息的目的地;消息发送给谁.Destination destination;// MessageProducer：消息发送者MessageProducer producer;// TextMessage message;// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jarconnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");try {// 构造从工厂得到连接对象connection = connectionFactory.createConnection();// 启动connection.start();// 获取操作连接session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);// 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置destination = session.createQueue("JiangQueue");// 得到消息生成者【发送者】producer = session.createProducer(destination);// 设置不持久化，此处学习，实际根据项目决定producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);// 构造消息，此处写死，项目就是参数，或者方法获取sendMessage(session, producer);session.commit();} catch (Exception e) {e.printStackTrace();} finally {try {if (null != connection)connection.close();} catch (Throwable ignore) {}}}public static void sendMessage(Session session, MessageProducer producer) throws Exception {for (int i = 1; i <= SEND_NUMBER; i++) {TextMessage message = session.createTextMessage("这是消息" + i);// 发送消息到目的地方System.out.println("发送消息：" + "这是消息" + i);producer.send(message);}}}

5. 接收端测试

package com.jiang.activemq;import javax.jms.Connection;import javax.jms.ConnectionFactory;import javax.jms.Destination;import javax.jms.MessageConsumer;import javax.jms.Session;import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;public class Receiver {    public static void main(String[] args) {        // ConnectionFactory ：连接工厂，JMS 用它创建连接        ConnectionFactory connectionFactory;        // Connection ：JMS 客户端到JMS Provider 的连接        Connection connection = null;        // Session： 一个发送或接收消息的线程        Session session;        // Destination ：消息的目的地;消息发送给谁.        Destination destination;        // 消费者，消息接收者        MessageConsumer consumer;        connectionFactory = new ActiveMQConnectionFactory(                ActiveMQConnection.DEFAULT_USER,                ActiveMQConnection.DEFAULT_PASSWORD,                "tcp://localhost:61616");        try {            // 构造从工厂得到连接对象            connection = connectionFactory.createConnection();            // 启动            connection.start();            // 获取操作连接            session = connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);            // 获取session注意参数值xingbo.xu-queue是一个服务器的queue，须在在ActiveMq的console配置            destination = session.createQueue("JiangQueue");            consumer = session.createConsumer(destination);            while (true) {                //设置接收者接收消息的时间，为了便于测试，这里谁定为100s                TextMessage message = (TextMessage) consumer.receive(100000);                if (null != message) {                    System.out.println("收到消息：" + message.getText());                } else {                    break;                }            }        } catch (Exception e) {            e.printStackTrace();        } finally {            try {                if (null != connection)                    connection.close();            } catch (Throwable ignore) {            }        }    }}

四、Queue与Topic的比较
1. JMS Queue执行load balancer语义
  一条消息仅能被一个consumer收到。如果在message发送的时候没有可用的consumer，那么它将被保存一直到能处理该message的consumer可用。如果一个consumer收到一条message后却不响应它，那么这条消息将被转到另一个consumer那儿。一个Queue可以有很多consumer，并且在多个可用的consumer中负载均衡。

2. Topic实现publish和subscribe语义
  一条消息被publish时，它将发到所有感兴趣的订阅者，所以零到多个subscriber将接收到消息的一个拷贝。但是在消息代理接收到消息时，只有激活订阅的subscriber能够获得消息的一个拷贝。

3.分别对应两种消息模式Point-to-Point (点对点),Publisher/Subscriber Model (发布/订阅者)
  其中在Publicher/Subscriber 模式下又有Nondurable subscription（非持久订阅）和durable subscription (持久化订阅)2种消息处理方式。