ActiveMQ 了解

 

Active MQ 是JMS的一个具体实现，所以首先要对JMS有所了解。

<!--[if !supportLists]-->1. <!--[endif]-->JMS:

定义：Java消息服务（Java Message Service）,是Sun是提出来的为J2EE提供企业消息处理的一套规范，他提供了创建，发送，接受，读取消息的服务。能接受消息生产者的（Message Producer）发出的消息，并把消息转发给消息消费者（Message Consumer）。（简单的说：Jms就是一个 处理消息发送 和接受的东东！）

<!--[if !supportLists]-->2. <!--[endif]-->JMS消息组成

JMS中的消息传递系统中传递的东东就是消息（Message）。其结构图如下：

<!--[if gte vml 1]><v:shapetype id="_x0000_t75" coordsize="21600,21600" o:spt="75" o:preferrelative="t" path="m@4@5l@4@11@9@11@9@5xe" filled="f" stroked="f"> <v:stroke joinstyle="miter" /> <v:formulas> <v:f eqn="if lineDrawn pixelLineWidth 0" /> <v:f eqn="sum @0 1 0" /> <v:f eqn="sum 0 0 @1" /> <v:f eqn="prod @2 1 2" /> <v:f eqn="prod @3 21600 pixelWidth" /> <v:f eqn="prod @3 21600 pixelHeight" /> <v:f eqn="sum @0 0 1" /> <v:f eqn="prod @6 1 2" /> <v:f eqn="prod @7 21600 pixelWidth" /> <v:f eqn="sum @8 21600 0" /> <v:f eqn="prod @7 21600 pixelHeight" /> <v:f eqn="sum @10 21600 0" /> </v:formulas> <v:path o:extrusionok="f" gradientshapeok="t" o:connecttype="rect" /> <o:lock v:ext="edit" aspectratio="t" /> </v:shapetype><v:shape id="_x0000_i1025" type="#_x0000_t75" style='width:274.5pt; height:337.5pt'> <v:imagedata src="file:///C:/DOCUME~1/ADMINI~1/LOCALS~1/Temp/msohtml1/01/clip_image001.png" o:title="" /> </v:shape><![endif]--><!--[if !vml]--><!--[endif]-->
Message 分为三个组成部分：

第一部分：
头（header）是个标准字段集，客户机和供应商都用它来标识和路由消息。

第二部分：
属性（property）支持把可选头字段添加到消息。如果您的应用程序需要不使用标准头字段对消息编目和分类，您就可以添加一个属性到消息以实现这个编目和分类。提供 set<Type>Property(...) 和 get<Type>Property(...) 方法以设置和获取各种 Java 类型的属性，包括 Object。JMS 定义了一个供应商选择提供的标准属性集。

第三部分：
消息的主体（body）包含要发送给接收应用程序的内容。每个消息接口特定于它所支持的内容类型。
JMS 为不同类型的内容提供了它们各自的消息类型，但是所有消息都派生自 Message 接口。
· StreamMessage：包含 Java 基本数值流，用标准流操作来顺序的填充和读取。
· MapMessage：包含一组名/值对；名称为 string 类型，而值为 Java 的基本类型。
· TextMessage：包含一个 String。
· ObjectMessage：包含一个 Serializable Java 对象；能使用 JDK 的集合类。
· BytesMessage：包含未解释字节流: 编码主体以匹配现存的消息格式。
· XMLMessage: 包含XML内容。扩展TextMessage,XMLMessage 类型的使用，使得消息过滤非常便利。

3. JMS的消息通信模型有两种：
JMS是用于和面向消息的中间件相互通信的应用程序接口。

（1）支持点对点(point-to-point)的域（简称PTP），

点对点消息传递域（PTP）的特点如下：

• 每个消息只能有一个消费者。

• 消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发

送消息的时候是否处于运行状态，它都可以提取消息。

（2）支持发布/订阅(publish/subscribe)类型的域（简称PUB/SUB）），

并且提供对下列类型的支持：经认可的消息传递,事务型消息的传递，一致性消息和具有持久性的订阅者支持。JMS还提供了另一种方式来对您的应用与旧的后台系统相集成。

发布/订阅消息传递域（PUB/SUB）的特点如下：

• 每个消息可以有多个消费者。

• 生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费

自它订阅之后发布的消息。JMS 规范允许客户创建持久订阅，这在一定程

度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激

活状态时发送的消息。

注意：在点对点（PTP）消息传递域中，目的地被成为队列（queue）；在发布/订阅（PUB/SUB）消息传递域中，目的地被成为主题（topic）。

queue和topic之间的详解图：

	Topic	Queue
概要	Publish Subscribe messaging 发布订阅消息	Point-to-Point 点对点
有无状态	topic数据默认不落地，是无状态的。	Queue数据默认会在mq服务器上以文件形式保存，比如Active MQ一般保存在$AMQ_HOME/data/kr-store/data下面。也可以配置成DB存储。
完整性保障	并不保证publisher发布的每条数据，Subscriber都能接受到。	Queue保证每条数据都能被receiver接收。
消息是否会丢失	一般来说publisher发布消息到某一个topic时，只有正在监听该topic地址的sub能够接收到消息；如果没有sub在监听，该topic就丢失了。（必须监听才能接收到消息）	Sender发送消息到目标Queue，receiver可以异步接收这个Queue上的消息。Queue上的消息如果暂时没有receiver来取，也不会丢失。
消息发布接收策略	一对多的消息发布接收策略，监听同一个topic地址的多个sub都能收到publisher发送的消息。Sub接收完通知mq服务器	一对一的消息发布接收策略，一个sender发送的消息，只能有一个receiver接收。receiver接收完后，通知mq服务器已接收，mq服务器对queue里的消息采取删除或其他操作。

4.消息发出去后的确认模式

应用程序创建的会话有一般有5 种确认模式（非事务）。

五种确认模式说明:
· AUTO_ACKNOWLEDGE：自动确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收。
· CLIENT_ACKNOWLEDGE：客户端确认模式。会话对象依赖于应用程序对被接收的消息调用一个acknowledge()方法。一旦这个方法被调用，会话会确认最后一次确认之后所有接收到的消息。这种模式允许应用程序以一个调用来接收，处理并确认一批消息。注意：在管理控制台中，如果连接工厂的Acknowledge Policy（确认方针）属性被设置为"Previous"（提前），但是你希望为一个给定的会话确认所有接收到的消息，那么就用最后一条消息来调用acknowledge()方法。
· DUPS_OK_ACKNOWLEDGE：允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。在需要考虑资源使用时，这种模式非常有效。注意：如果你的应用程序无法处理重复的消息的话，你应该避免使用这种模式。如果发送消息的初始化尝试失败，那么重复的消息可以被重新发送。
· NO_ACKNOWLEDGE：不确认模式。不确认收到的消息是需要的。消息发送给一个NO_ACKNOWLEDGE 会话后，它们会被WebLogic 服务器立即删除。在这种模式下，将无法重新获得已接收的消息，而且可能导致下面的结果：1. 消息可能丢失；和（或者）另一种情况：2. 如果发送消息的初始化尝试失败，会出现重复消息被发送的情况。
· MULTICAST_NO_ACKNOWLEDGE：IP组播下的不确认模式，同样无需确认。发送给一个MULTICAST_NO_ACKNOWLEDGE会话的消息， 会共享之前所述的NO_ACKNOWLEDGE 确认模式一样的特征。这种模式支持希望通过IP 组播方式进行消息通信的应用程序，而且无需依赖会话确认提供的服务质量。注意：如果你的应用程序无法处理消息的丢失或者重复，那么你应该避免使用这种模式。如果发送消息的初始化尝试失败的话，重复的消息可能会被再次发送。
（注：在上表的5 种确认模式中，AUTO_ACKNOWLEDGE ，DUPS_OK_ACKNOWLEDGE 和
CLIENT_ACKNOWLEDGE 是JMS 规范定义的，NO_ACKNOWLEDGE 和MULTICAST_NO_ACKNOWLEDGE是WebLogic JMS 提供的。）

5． JMS 支持以下两种消息提交模式：

. PERSISTENT。指示 JMS provider 持久保存消息，以保证消息不会因为 JMS

provider 的失败而丢失。（当服务器重启之后，之前的发送PERSISTENT消息，会获取到！）

.NON_PERSISTENT。不要求 JMS provider 持久保存消息。（当服务器重启之后，之前的发送NON_PERSISTENT消息，会获取不到！）

 

6.一种典型的JMS 程序需要经过下列几个步骤:
· 通过 JNDI 查找 ConnectionFactory。
· 用 ConnectionFactory 创建一个 Connection。
· 用 Connection 创建一个或多个 Session。
· 用 Session 和 Destination 创建所需的 MessageProducer 和 MessageConsumer。
· 启动 Connection。(下面是对应的截图说明)

<!--[if gte vml 1]><v:shape id="_x0000_i1026" type="#_x0000_t75" alt="JMS编程模型" style='width:298.5pt;height:223.5pt'> <v:imagedata src="file:///C:/DOCUME~1/ADMINI~1/LOCALS~1/Temp/msohtml1/01/clip_image003.jpg" o:href="http://infocenter.apusic.com/help/topic/amq/v6/img/mq/jms-program-model.jpg" /> </v:shape><![endif]--><!--[if !vml]--><!--[endif]-->

JMS 各个步骤的详细介绍

先简单介绍JMS的服务器端

（0）Broker ：可以把JMS Brokers看做服务器端，这个服务器可以独立运行，也可以随着其他容器，并且它是以内嵌的方式运行，如下配置

BrokerService broker = new BrokerService();

broker.addConnector("tcp://localhost:61616");

broker.start();

此外，也可以通过BrokerFactory来创建broker，例如：

BrokerService broker = BrokerFactory.createBroker(new URI(someU

RI));

通过ActiveMQConnectionFactory还可以隐含创建内嵌的broker（后面的实例代码会用到）

下面正正式介绍每一个步骤

(1)连接工厂 (ConnectionFactory)

连接工厂是客户用来创建连接的对象.

例如: ActiveMQ 提供的ActiveMQConnectionFactory。注意（要初始化 JMS，则需要使用连接工厂。客户端通过创建 ConnectionFactory建立到 ActveMQ的连接，一个连接工厂封装了一组连接配置参数，这组参数在配置ActiveMQ时已经定义，例如brokerURL参数，此参数传入的是ActiveMQ服务地址和端口，支持openwire协议的默认连接为 tcp://localhost:61616，支持 stomp协议的默认连接为tcp://localhost:61613。
ActiveMQConnectionFactory构造方法：

ActiveMQConnectionFactory();

ActiveMQConnectionFactory(String brokerURL);

ActiveMQConnectionFactory(String userName, String password, String b rokerURL) ; ActiveMQConnectionFactory(String userName, String password, URI brok erURL) ; ActiveMQConnectionFactory(URI brokerURL); 其中 brokerURL为ActiveMQ服务地址和端口。
例如：
　　ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192. 168.0.135:61616");
　　或者
　　ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
　　connectionFactory. setBrokerURL("tcp://192.168.0.135:61616");）

(2)连接 (Connection)

Connection 封装了客户与 JMS 提供者之间的一个虚拟的连接。

注意

（当一个Connection被创建时，它的传输默认是关闭的，必须使用start方法开启。一个Connection可以建立一个或多个的Session。当一个程序执行完成后，必须关闭之前创建的Connection，否则 ActiveMQ不能释放资源，关闭一个Connection同样也关闭了 Session，MessageProducer和MessageConsumer。）

（3）会话(Seesion)

Session 是生产和消费消息的一个单线程上下文。

会话用于创建消息生产者（producer）、消息消费者（consumer）和消息（message）等。会话提供了一个事务性的上下文，在这个上下文中，一组发送和接收被组合到了一个原子操作中。

（4）目的地

目的地是客户用来指定它生产的消息的目标和它消费的消息的来源的对象。

JMS1.0.2 规范中定义了两种消息传递域：点对点（PTP）消息传递域和发布/订阅

消息传递域。（关于PTP和PUB/SUB 在“JMS的消息通信模型有两种“ 有详细介绍！）

消息生产者（producer）：用于把消息发送到一个目的地。

消息消费者（consumer）：它用于接收发送到目的地的消息。

消息的消费可以采用以下两种方法之一：

• 同步消费。通过调用消费者的 receive 方法从目的地中显式提取消息。

receive 方法可以一直阻塞到消息到达。

• 异步消费。客户可以为消费者注册一个消息监听器（Listener），以定义在消息到达时所采取的动作。

 

7．现在进行简单的实例的配置和测试

Active MQ配置和启动：

比如我下的是最新的包apache-activemq-5.4.2-bin.zip

1、安装文件：apache-activemq-5.4.2-bin.zip

　　2、安装过程：解压缩到apache-activemq-5.4.2-bin.zip到一个目录，比如C:/Program Files/apache-activemq-5.4.2

　　3、ActiveMQ配置文件在C:/Program Files/apache-activemq-5.4.2/conf/activemq.xml,采用默认的配置就可以运行了，多台做集群时可以修改该文件。

　　4、启动ActiveMQ：运行C:/Program Files/apache-activemq-5.4.2/bin/activemq.bat

　　5、测试

　　ActiveMQ默认使用的TCP连接端口是61616, 通过查看该端口的信息可以测试ActiveMQ是否进入DOS:netstat -an|find "61616"

　　TCP 0.0.0.0:61616 0.0.0.0:0 LISTENING

　　6、监控

　　ActiveMQ5.0版本默认启动时，启动了内置的jetty服务器，提供一个demo应用和用于监控ActiveMQ的admin应用。

　　demo：http://127.0.0.1:8161/demo/
下面是一个简单的例子（一个简单的实例）

我先建一个简单的Java Project 项目（项目名称activemq），

并且将我自己的安装目录C:/Program Files/apache-activemq-5.4.2下的activemq-all-5.4.2.jar的包加入到Reference Libraries中，这个包包含了所有jms接口api的实现。

消息的生产者ProducerTool类

import javax.jms.Connection;

import javax.jms.DeliveryMode;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MessageProducer;

import javax.jms.Session;

import javax.jms.TextMessage;

import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ProducerTool {

private String user = ActiveMQConnection.DEFAULT_USER; //默认 用户

private String password = ActiveMQConnection.DEFAULT_PASSWORD; //默认 密码

private String url = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的是localhost:8080

private String subject = "TOOL.DEFAULT"; //消息目的地名称

private Destination destination = null; //在点对点（PTP）消息传递域中，目的地被成为队列（queue）

private Topic topic=null;//在发布/订阅（PUB/SUB）消息传递域中，目的地被成为主题（topic）。

private Connection connection = null; //初始化 一个JMS客户端到JMS Provider的连接

private Session session = null; //初始化 一个发送消息的进程

private MessageProducer producer = null; //初始化 消息生产者 (它是由session 创建的)

// 初始化

private void initialize() throws JMSException, Exception {

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(

user, password, url);

connection = connectionFactory.createConnection();

//false 参数表示 为非事务型消息，后面的参数表示消息的确认类型（见4.消息发出去后的确认模式）

session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//PTP消息方式 目的地被成为队列（queue）

destination = session.createQueue(subject);

producer = session.createProducer(destination);

//在发布/订阅（PUB/SUB）消息，目的地被成为主题（topic）。

/*topic=session.createTopic(subject);

producer=session.createProducer(topic);

*/

//消息模型为非持久型

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

}

// 发送消息

public void produceMessage(String message) throws JMSException, Exception {

initialize();

TextMessage msg = session.createTextMessage(message);

connection.start();

System.out.println("Producer:->Sending message: " + message);

producer.send(msg);

System.out.println("Producer:->Message sent complete!");

}

// 关闭连接

public void close() throws JMSException {

System.out.println("Producer:->Closing connection");

if (producer != null)

producer.close();

if (session != null)

session.close();

if (connection != null)

connection.close();

}

}

消息的消费者ConsumerTool类
import javax.jms.Connection;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MessageConsumer;

import javax.jms.Session;

import javax.jms.MessageListener;

import javax.jms.Message;

import javax.jms.TextMessage;

import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ConsumerTool implements MessageListener {

private String user = ActiveMQConnection.DEFAULT_USER; //默认 用户

private String password = ActiveMQConnection.DEFAULT_PASSWORD; //默认 密码

private String url = ActiveMQConnection.DEFAULT_BROKER_URL; //默认的是localhost:8080

private String subject = "TOOL.DEFAULT"; //消息目的地名称

private Destination destination = null; //在点对点（PTP）消息传递域中，目的地被成为队列（queue）

private Topic topic=null;//在发布/订阅（PUB/SUB）消息传递域中，目的地被成为主题（topic）。

private Connection connection = null; //初始化 一个JMS客户端到JMS Provider的连接

private Session session = null; //初始化 一个接受消息的进程

private MessageConsumer consumer = null; //初始化 消息消费者

// 初始化

private void initialize() throws JMSException, Exception {

ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(

user, password, url);

connection = connectionFactory.createConnection();

//false 参数表示 为非事务型消息，后面的参数表示消息的确认类型（见4.消息发出去后的确认模式）

session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

//PTP消息方式 目的地被成为队列（queue）

destination = session.createQueue(subject);

consumer = session.createConsumer(destination);

//在发布/订阅（PUB/SUB）消息，目的地被成为主题（topic）。

/*topic=session.createTopic(subject);

consumer = session.createConsumer(topic);

*/

}

// 消费消息

public void consumeMessage() throws JMSException, Exception {

initialize();

connection.start();

System.out.println("Consumer:->Begin listening...");

//接受消息方式1：消息的异步接收 监听 实现MessageListener接口，每当消息到达时，ActiveMQ会调用MessageListener中的onMessage函数。

consumer.setMessageListener(this);

//接受消息方式2：消息的同步 主动接受消息 用recieve方式

/* TextMessage textMessage = (TextMessage)consumer.receive();

if(null!=textMessage)

System.out.println(textMessage.getText());

else

System.out.println("无"); */

}

// 关闭连接

public void close() throws JMSException {

System.out.println("Consumer:->Closing connection");

if (consumer != null)

consumer.close();

if (session != null)

session.close();

if (connection != null)

connection.close();

}

// 消息处理函数

public void onMessage(Message message) {

try {

if (message instanceof TextMessage) {

TextMessage txtMsg = (TextMessage) message;

String msg = txtMsg.getText();

System.out.println("Consumer:->Received: " + msg);

} else {

System.out.println("Consumer:->Received: " + message);

}

} catch (JMSException e) {

e.printStackTrace();

}

}

}

test测试类

import javax.jms.JMSException;

public class test {

/**

* @param args

*/

public static void main(String[] args) throws JMSException, Exception {

ConsumerTool consumer = new ConsumerTool();

ProducerTool producer = new ProducerTool();

//接受消息方式一：通过监听Listener 开始监听

consumer.consumeMessage();

// 延时500毫秒之后发送消息

Thread.sleep(500);

producer.produceMessage("Hello, world!");

//接受消息方式2：主动方式， 必须在producer产生消息后，去获取，否则获取不到

//consumer.consumeMessage();

producer.close();

// 延时500毫秒之后停止接受消息

Thread.sleep(500);

consumer.close();

}

}

/*消息的同步与异步接收

　消息的同步接收是指客户端主动去接收消息，客户端可以采用MessageConsumer的receive方法去接收下一个消息。

　消息的异步接收是指当消息到达时，ActiveMQ主动通知客户端。客户端可以通过注册一个实现 MessageListener接口的对象到MessageConsumer。

MessageListener只有一个必须实现的方法onMessage，它只接收一个参数，即Message。

在为每个发送到Destination的消息实现onMessage时，将调用该方法。

Message receive() Message receive(long timeout) Message receiveNoWait()

其中timeout为等待时间，单位为毫秒。*/

先启动ActiveMQ；

运行test类的运行结果

<!--[if gte vml 1]><v:shape id="_x0000_i1027" type="#_x0000_t75" style='width:414.75pt;height:100.5pt'> <v:imagedata src="file:///C:/DOCUME~1/ADMINI~1/LOCALS~1/Temp/msohtml1/01/clip_image004.png" o:title="" /> </v:shape><![endif]--><!--[if !vml]--><!--[endif]-->

在学习ActiveMQ的过程中要注意的地方：

1.注意activeMQ 内含了jefft 服务器，所以要在运行activeMQ实例之前要，启动activemq.bat.
2.主要引用的的包activemq-all-5.4.2.jar 就可以了，他包jms所需要的接口。

3.对于消息topic模式，接受消息的那一方，要使用监听方式，获取消息，否则获取不到。

未解决：

tomcat 与activemq整合失败，正在尝试.

(activemq官方也有个HelloWorld例子，http://activemq.apache.org/hello-world.html)