ActiveMQ入门实例

ActiveMQ 是Apache出品的开源消息总线。

全然支持JMS1.1规范

　　首先我们要了解一下JMS

JMS简单介绍

Java消息服务（Java Message Service，JMS）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API。用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与详细平台无关的API。绝大多数MOM提供商都对JMS提供支持。

Java消息服务的规范包含两种消息模式，点对点和公布者／订阅者。很多提供商支持这一通用框架因此，程序猿能够在他们的分布式软件中实现面向消息的操作，这些操作将具有不同面向消息中间件产品的可移植性。

Java消息服务支持同步和异步的消息处理。在某些场景下，异步消息是必要的；在其它场景下，异步消息比同步消息操作更加便利。

Java消息服务支持面向事件的方法接收消息。事件驱动的程序设计如今被广泛觉得是一种富有成效的程序设计范例

在应用系统开发时。Java消息服务能够推迟选择面对消息中间件产品，也能够在不同的面对消息中间件切换。

JMS消息格式：

JMS定义了五种不同的消息正文格式，以及调用的消息类型。同意你发送并接收以一些不同形式的数据。提供现有消息格式的一些级别的兼容性。
· StreamMessage Java原始值的数据流
· MapMessage 一套名称-值对
· TextMessage 一个字符串对象
· ObjectMessage 一个序列化的 Java对象
· BytesMessage 一个未解释字节的数据流

JMS元素：

JMS由下面7个元素组成

1.JMS提供者(JMS provider)

　　连接面向消息中间件的。JMS接口的一个实现。提供者能够是Java平台的JMS实现。也能够是非Java平台的面向消息中间件的适配器。

2. JMS客户（JMS client）

　　生产或消费消息的基于Java的应用程序或对象。

3. JMS生产者(JMS producer/publisher)

　　创建并发送消息的JMS客户。

4. JMS消费者(JMS consumer/subscriber)

　　接收消息的JMS客户。

5. JMS消息(JMS message)

　　包含能够在JMS客户之间传递的数据的对象

6. JMS队列(JMS queue)

　　一个容纳那些被发送的等待阅读的消息的区域。

队列暗示，这些消息将依照顺序发送。

一旦一个消息被阅读，该消息将被从队列中移走。

7. JMS主题(JMS topic)

　　一种支持发送消息给多个订阅者的机制。

JMS消息传输方式：

Java消息服务应用程序结构支持两种模型：

点对点或队列模型(Point-to-point model)

　　一个生产者向一个特定的队列公布消息，一个消费者从该队列中读取消息。

这里，生产者知道消费者的队列，并直接将消息发送到消费者的队列。这样的模式被概括为：

　　· 仅仅有一个消费者将获得消息

　　· 生产者不须要在接收者消费该消息期间处于执行状态，接收者也相同不须要在消息发送时处于执行状态。

　　· 每个成功处理的消息都由接收者签收

公布者／订阅者模型(Publish/subscribe model)

　　支持向一个特定的消息主题公布消息。0或多个订阅者可能对接收来自特定消息主题的消息感兴趣。在这样的模型下，公布者和订阅者彼此不知道对方。这样的模式好比是匿名公告板。这样的模式被概括为：

　　· 多个消费者能够获得消息

　　· 在公布者和订阅者之间存在时间依赖性。

公布者须要建立一个订阅（subscription），以便客户可以购订阅。订阅者必须保持持续的活动状态以接收消息。除非订阅者建立了持久的订阅。

在那种情况下。在订阅者未连接时公布的消息将在订阅者又一次连接时又一次公布。

使用Java语言，JMS提供了将应用与提供数据的传输层相分离的方式。同一组Java类能够通过JNDI中关于提供者的信息。连接不同的JMS提供者。

这一组类首先使用一个连接工厂以连接到队列或主题，然后发送或公布消息。在接收端，客户接收或订阅这些消息。

JMS应用程序接口

Java消息服务的API在javax.jms(J2EE)包中提供。

ConnectionFactory 接口（连接工厂）

用户用来创建到JMS提供者的连接的被管对象。JMS客户通过可移植的接口訪问连接，这样当下层的实现改变时。代码不须要进行改动。管理员在JNDI名字空间中配置连接工厂，这样，JMS客户才可以查找到它们。依据消息类型的不同，用户将使用队列连接工厂。或者主题连接工厂。

Connection 接口（连接）

连接代表了应用程序和消息server之间的通信链路。在获得了连接工厂后。就能够创建一个与JMS提供者的连接。依据不同的连接类型，连接同意用户创建会话，以发送和接收队列和主题到目标。

Destination 接口（目标）

目标是一个包装了消息目标标识符的被管对象。消息目标是指消息公布和接收的地点。或者是队列，或者是主题。

JMS管理员创建这些对象，然后用户通过JNDI发现它们。

和连接工厂一样，管理员能够创建两种类型的目标，点对点模型的队列。以及公布者／订阅者模型的主题。

MessageConsumer 接口（消息消费者）

由会话创建的对象。用于接收发送到目标的消息。

消费者能够同步地（堵塞模式）。或异步（非堵塞）接收队列和主题类型的消息。

MessageProducer 接口（消息生产者）

由会话创建的对象。用于发送消息到目标。

用户能够创建某个目标的发送者。也能够创建一个通用的发送者。在发送消息时指定目标。

Message 接口（消息）

是在消费者和生产者之间传送的对象，也就是说从一个应用程序创送到还有一个应用程序。一个消息有三个主要部分：

　　1. 消息头（必须）：包括用于识别和为消息寻找路由的操作设置。

　　2. 一组消息属性（可选）：包括额外的属性，支持其它提供者和用户的兼容。能够创建定制的字段和过滤器（消息选择器）。

　　3. 一个消息体（可选）：同意用户创建五种类型的消息（文本消息，映射消息。字节消息。流消息和对象消息）。

消息接口很灵活，并提供了很多方式来定制消息的内容。

Session 接口（会话）

表示一个单线程的上下文。用于发送和接收消息。因为会话是单线程的。所以消息是连续的。就是说消息是依照发送的顺序一个一个接收的。

会话的优点是它支持事务。

假设用户选择了事务支持，会话上下文将保存一组消息，直到事务被提交才发送这些消息。在提交事务之前，用户能够使用回滚操作取消这些消息。一个会话同意用户创建消息生产者来发送消息，创建消息消费者来接收消息。

在早期版本号的 JMS 中，必须使用的 pub/sub 和点对点编程模型具有不同的类层次结构。保留这些类层次结构是为了支持与早期版本号的 JMS API 的向后兼容，但鼓舞client开发者使用通用接口。

PTP 与 Pub/Sub 接口之间的关系

JMS 通用	PTP 域	Pub/Sub 域
ConnectionFactory	QueueConnectionFactory	TopicConnectionFactory
Connection	QueueConnection	TopicConnection
Destination	Queue	Topic
Session	QueueSession	TopicSession
MessageProducer	QueueSender	TopicPublisher
MessageConsumer	QueueReceiver	TopicSubscriber

以下对这些 JMS 概念进行简单定义。有关很多其它信息，请參阅 JMS 规范的 PTP 和 Pub/Sub 章节。

ConnectionFactory - client用于创建 Connection 的管理对象
Connection - 到 JMS 提供者的活动连接
Destination - 封装消息目的地标识的管理对象
Session - 发送和接收消息的单线程上下文
MessageProducer - Session 创建的对象，用于将消息发送到目的地
MessageConsumer - Session 创建的对象，用于接收发送到目的地的消息

使用ActiveMQ

部署ActiveMQ

首先我们能够到官方下载页去下载最新的ActiveMQ的部署程序(http://activemq.apache.org/download-archives.html)

将程序解压后假设已经配置了java环境能够直接执行安装文件夹/bin下的activemq.bat来启动activemq程序，系统会自己主动执行启动过程。当然一般安装失败的情况是没有装JVM环境，启动成功应该是这样

activemq使用了jettyserver来进行管理，能够在conf/jetty.xml文件里对其配置，conf/activemq.xml文件里对activemq进行配置

打开浏览器输入http://localhost:8161/admin/默认配置是这个，当然你也能够更改这个配置

在Java项目中使用activemq

在javaproject中导入ActiveMQ须要的包

须要例如以下包：

　　· activemq-core.jar

　　· activeio-core.jar

　　· *kahadb.jar (if you wish to use persistence,假设要使用持久化须要此jar包)

　　· slf4j-api.jar

　　· J2EE Jars

geronimo-spec-jms.jar
geronimo-spec-jta.jar
geronimo-spec-j2ee-management.jar

也能够使用默认的activemq-all.jar，下载地址http://mvnrepository.com/artifact/org.apache.activemq/activemq-all)

生产消息过程

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 client到JMS Provider 的连接
        Connection connection = null;
        // Session： 一个发送或接收消息的线程
        Session session;
        // Destination ：消息的目的地;消息发送给谁.
        Destination destination;
        // MessageProducer：消息发送者
        MessageProducer producer;
        // TextMessage message;
        // 构造ConnectionFactory实例对象。此处採用ActiveMq的实现jar
        connectionFactory = 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是一个server的queue，须在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            // 得到消息生成者【发送者】
            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("ActiveMq 发送的消息" + i);
            // 发送消息到目的地方
            System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
            producer.send(message);
        }
    }
}

消费消息过程

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 client到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是一个server的queue。须在在ActiveMq的console配置
            destination = session.createQueue("FirstQueue");
            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) {
            }
        }
    }
}

消息消费者有两种消息接收方式

　　1. 上面演示样例中的调用receive方法。使用堵塞的方式获取消息

　　2. 使用MessageConsumer对象的setMessageListener方法设置监听的方式，设置监听后假设关闭Consumer、session或connection将终止监听

JMS消息类型

JMS API 定义了五种消息正文类型：

Stream（流）- StreamMessage 对象的消息正文包括 Java 编程语言原始值流（“Java 基本类型”）。
按顺序填充和读取。
Map（映射）- MapMessage 对象的消息正文包括一组名称-值对，当中名称是 String 对象，值是 Java 基本类型。
能够依据名称按顺序或随机訪问这些条目。条目的顺序是不确定的。
Text（文本）- TextMessage 对象的消息正文包括 java.lang.String 对象。此消息类型可用于传输纯文本消息和 XML 消息。
Object（对象）- ObjectMessage 对象的消息正文包括 Serializable Java 对象。
Byte（字节）- BytesMessage 对象的消息正文包括未解释的字节流。此消息类型能够按字面意义编码正文，以匹配现有的消息格式。在大多数情况下，能够使用更易用的其它正文类型。
虽然 JMS API 同意将消息属性用于字节消息，但一般不使用它们，由于包括属性可能会影响格式。

ACK_MODE确认模式

确认模式用来指示会话怎样确认收到消息。在使用Connection的createSession方法时须要指定此选项，在JMS的Session接口中包括以下四个选项

　　· AUTO_ACKNOWLEDGE = 1 自己主动确认

　　当会话从对 receive 的调用成功返回时，或在会话已调用的用于处理消息的消息侦听器成功返回时，会话会自己主动确认client的消息接收。

　　· CLIENT_ACKNOWLEDGE = 2 client手动确认

　　通过此确认模式，client通过调用消息的 acknowledge 方法确认已使用的消息。确认已使用的消息将确认该会话已使用的全部消息。

　　· DUPS_OK_ACKNOWLEDGE = 3 自己主动批量确认

　　此确认模式指示会话延迟确认消息的传送。这可能在 JMS 提供者失败的情况下导致传送某些反复消息。因此仅仅有能同意反复消息的使用方才应使用此模式。

使用此模式能够通过最大限度地降低会话为防止反复所做的工作。从而降低会话开销。

　　· SESSION_TRANSACTED = 0 事务提交并确认

　　假设会话是事务的则使用此模式，忽略设置的其它模式值

　　在事务开启之后，和session.commit()之前，所有消费的消息。要么所有正常确认。要么所有redelivery。这样的严谨性，通常在基于GROUP(消息分组)或者其它场景下特别适合。