ActiveMQ入门

  categories:mq, 资料  tags:,   author:

摘要:本文主要讲述ActiveMQ的基本知识和使用方法,并简单结合spring使用ActiveMQ

一、ActiveMQ特性和使用总览

企业消息软件从80年代起就存在,它不只是一种应用间消息传递风格,也是一种集成风格。因此,消息传递可以满足应用间的通知和互相操作。但是开源的解决方案是到最近10年才出现的。Apache ActiveMQ就是其中一种。它使应用间能以异步,松耦合方式交流。本章将向您介绍ActiveMQ

ActiveMQApache软件基金下的一个开源软件,它遵循JMS1.1规范(Java Message Service),是消息驱动中间件软件(MOM)。它为企业消息传递提供高可用,出色性能,可扩展,稳定和安全保障。ActiveMQ使用Apache许可协议。因此,任何人都可以使用和修改它而不必反馈任何改变。这对于商业上将ActiveMQ用在重要用途的人尤为关键。MOM的工作是在分布式的各应用之间调度事件和消息,使之到达指定的接收者。所以高可用,高性能,高可扩展性尤为关键。

ActiveMQ的目标是在尽可能多的平台和语言上提供一个标准的,消息驱动的应用集成。ActiveMQ实现JMS规范并在此之上提供大量额外的特性。

下面是一个高层次的特性列表。

·遵循JMS规范 —-理解ActiveMQ的起始点是明白ActiveMQ的各种特性是JMS1.1规范的实现。本章后面将讨论JMS规范提供的好处和保证。它们包括同步和异步消息传递,一次和只有一次的消息传递,对于预订者的持久消息等等。依附于JMS规范意味着,不论JMS消息提供者是谁,同样的基本特性都是有效的。

·连接—-ActiveMQ提供各种连接选择,包括HTTPHTTPSIP多点传送,SSLSTOMPTCPUDPXMPP等。大量的连接协议支持使之具有更好的灵活性。很多现有的系统使用一种特定协议并且不能改变,所以一个支持多种协议的消息平台降低了使用的门槛。虽然连接很重要,但是和其他容器集成也同样重要。第四章将讲解 ActiveMQ的传输连接器(transport connectors)和网络连接器(network connectors)。

·可插拔的持久性和安全—-ActiveMQ提供多种持久性方案可供选择,也可以完全按自己需求定制验证和授权。例如,ActiveMQ通过KahaDB提供自己的超快速消息持久方案(ultra-fast message persistence),但也支持标准的JDBC方案。ActiveMQ可以通过配置文件提供简单的验证和授权,也提供标准的JAAS登陆模块。

·用Java建立消息驱动应用—-ActiveMQ最常用在Java应用中,用于发送和接收消息。这部分的内容涉及JMS规范API

·与应用服务器集成—-ActiveMQjava应用服务器集成是很常见的。

·客户端APIs—-ActiveMQ对多种语言提供客户端API,除了Java之外还有C/C++.NETPerlPHPPythonRuby等。这使得ActiveMQ能用在Java之外的其它语言中。很多其它语言都可以通过ActiveMQ提供的客户端API使用ActiveMQ的全部特性。当然,ActiveMQ代理器(broker)仍然是运行在java虚拟机上,但是客户端能够使用其它的被支持的语言。

·代理器集群(Broker clustering—-为了利于扩展,多个ActiveMQ broker能够联合工作。这个方式就是network of brokers并且能支持多种拓扑结构。

·高级代理器特性和客户端选项—-ActiveMQ为代理器和客户端连接提供很多高级的特性。ActiveMQ也可以通过代理器的XML配置文件支持Apache Camel

·简单的管理—-ActiveMQ是为开发者设计的。它并不需要专门的管理工具,因为它提供各种易用且强大的管理特性。有很多方法去监控ActiveMQ的各个方面,可以通过JMX使用JConsoleActiveMQ web console;可以运行ActiveMQ消息报告;可以用命令行脚本;可以通过日志。

用 ActiveMQ 最好还是了解下 JMS

JMS 公共

点对点域

发布/订阅域

ConnectionFactory

QueueConnectionFactory

TopicConnectionFactory

Connection

QueueConnection

TopicConnection

Destination

Queue

Topic

Session

QueueSession

TopicSession

MessageProducer

QueueSender

TopicPublisher

MessageConsumer

QueueReceiver

TopicSubscriber

JMS 定义了两种方式:Quere(点对点);Topic(发布/订阅)。

·ConnectionFactory 是连接工厂,负责创建Connection。

·Connection 负责创建 Session。

·Session 创建 MessageProducer(用来发消息) 和 MessageConsumer(用来接收消息)。

·Destination 是消息的目的地。

详细的可以网上找些 JMS 规范(有中文版)

二、开始使用ActiveMQ

开始使用ActiveMQ并不难。你只要启动代理器并确保它能接受连接和发送消息。ActiveMQ有一些自带的例子。

在这部分,你将下载和安装Java SE,下载和安装ActiveMQ,检查ActiveMQ目录,然后第一次启动ActiveMQ。所需工具包括JDK1.5+AntActiveMQ

1 JDK下载和安装

ActiveMQ 要求Sun Java SE 1.5或以上。在开始这部分前,必须先安装。JDK的下载和安装就不是本文介绍重点。

2 Ant下载和安装

Ant可以用来构建和运行ActiveMQ自带例子。Ant可以从Apache Ant网址下载。URL: http://ant.apache.org/bindownload.cgi

点击链接地址并选择正确的压缩包。(tar包是LinuxUnixzipWindows)。请按照下列地址安装AntURL: http://ant.apache.org/manual/install.html。确保你设置好$ANT_HOME环境变量,并将$ANT_HOME/bin放到$PATH环境变量里。安装完毕后你可以运行下面的命令查看Ant版本。

你的可能跟我的不太一样,使用Ant的不同版本,不过这没关系。一旦Ant输出如上信息,你就可以确定Ant都安装正确。

3 ActiveMQ下载和安装

ActiveMQ可以从Apache ActiveMQ网站下,URL: http://activemq.apache.org/download.html.

点击地址到5.4.3版本,你可以看到tarzip格式包。(tar包是LinuxUnixzipWindows)。下载完后解压。我的解压到了E:\tools\apache-activemq-5.4.3

从命令行进入apache-activemq-5.4.3 目录,输入如下命令。

· LICENSE—-Apache Software Foundation(ASF)要求的一个文件.包含ActiveMQ使用的所有库的许可证.

· NOTICE—-ASF要求的另一个文件.包含ActiveMQ使用的所有库的版权信息.

· README.txt 一个包含一些URL的文档,使新手可以使用ActiveMQ.

· WebConsole-README.txt—- 包含使用ActiveMQ web console使用说明.

· activemq-all-5.4. 3.jar—一个jar包包含ActiveMQ所有东西。放在这里是方便你使用它。

· bin—-包含二进制或可运行文件。ActiveMQ启动脚本就放在里面。

· conf–ActiveMQ所有的配置信息。

· data–日志和持久化文件存储地方。

· docs–包含一个简单的index.html,该文件指向ActiveMQ网站。

· example—-ActiveMQ 例子。我们用这些例子来简单的测试ActiveMQ

· lib—-所有ActiveMQ所需库。

· user-guide.html—- 一个简单指引启动ActiveMQ和运行例子。

· webapps—-ActiveMQ web console和一些网络演示。

下一部分将启动 ActiveMQ并用这些例子验证它。

启动ActiveMQ

在命令行中输入一下命令,或者直接运行bin目录下的activemq.bat文件

刚才的命令启动了ActiveMQ代理器和一些连接器,使得客户端可以通过一些诸如TCPSSLSTOMPXMPP协议连接进来。请注意现在ActiveMQ已经启动,并且客户端可以通过TCP 61616端口连接进来。最好的方法是使用ActiveMQ自带的例子来发送和接收消息。下面我们来运行第一个例子。

4 、运行ActiveMQ第一个例子

这个例子是模拟生产者消费者的例子,生产者产生2000条信息,然后关闭,消费者消费2000信息,然后关闭。下面我们演示一下。

重新打开一个DOS窗口,切到MQ目录的example目录下,我的是E:\tools\apache-activemq-5.4.3\example

然后输入一下命令

最后一行显示程序在等待2000message。这是一个消费者。再重新打开DOS窗口并且到MQ目录下的example目录,输入一下命令ant producer,这个是生产者。输入命令之后你回看到刚才打开的两个DOS窗口打印出一堆东西

看到BUILD SUCCESSFUL说明程序运行成功。

消费者窗口变成了如下内容:

示例程序演示完毕,观看别人写的看着没啥意思,我们自己写一个程序,测试一下。

三、运行自己编写的第一个例子

我们自己编写一个跟上面类似的例子。

1 、建立web项目

打开myeclipse,新建web项目,内容填下如下:

点击finish。接着在项目根目录下新建一个lib目录,然后将我们上面提及的activemq-all-5.4.3.jar 拷贝至lib目录下,并将其加入项目library中如下图:

点击OK,项目目前的文件夹结构如下:

2 、编写代码

src上右键选择新建class,具体如下:

点击finish。然后输入如下代码

package com.mq;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.MessageProducer;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Producer {

public static void main(String[] args) {

String user = ActiveMQConnection. DEFAULT_USER;

String password = ActiveMQConnection. DEFAULT_PASSWORD;

String url = ActiveMQConnection. DEFAULT_BROKER_URL;

String subject = “TOOL.DEFAULT” ;

ConnectionFactory contectionFactory = new ActiveMQConnectionFactory(user,password,url);

try {

Connection connection = contectionFactory.createConnection();

connection.start();

Session session = connection.createSession(Boolean. TRUE, Session. AUTO_ACKNOWLEDGE);

Destination destination = session.createQueue(subject);

MessageProducer producer = session.createProducer(destination);

for( int i = 0;i<=20;i++){

MapMessage message = session.createMapMessage();

message.setLong( “count”, new Date().getTime());

Thread. sleep(1000);

producer.send(message);

System. out.println( “–发送消息:+ new Date());

}

session.commit();

session.close();

connection.close();

} catch (JMSException e) {

e.printStackTrace();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

然后再新建一个消费者类如下图:

点击finish。然后输入如下代码

package com.mq;

import java.util.Date;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.MessageConsumer;

import javax.jms.MessageListener;

import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {

public static void main(String[] args) {

String user = ActiveMQConnection. DEFAULT_USER;

String password = ActiveMQConnection. DEFAULT_PASSWORD;

String url = ActiveMQConnection. DEFAULT_BROKER_URL;

String subject = “TOOL.DEFAULT” ;

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);

Connection connection;

try {

connection = connectionFactory.createConnection();

connection.start();

final Session session = connection.createSession(Boolean. TRUE, Session. AUTO_ACKNOWLEDGE);

Destination destination = session.createQueue(subject);

MessageConsumer message = session.createConsumer(destination);

message.setMessageListener( new MessageListener(){

public void onMessage(Message msg) {

MapMessage message = (MapMessage)msg;

try {

System. out.println( “–收到消息:+ new Date(message.getLong(“count” )));

session.commit();

} catch (JMSException e) {

e.printStackTrace();

}

}

});

Thread. sleep(30000);

session.close();

connection.close();

} catch (JMSException e) {

e.printStackTrace();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

3 、运行程序

右键运行程序Consumer类,控制台打印出如下信息:

再右键运行Producer类,控制台上打印出来的信息同上。稍等一秒,Producer类的运行控制台会打印出来如下信息:

有信息可知,生产者正在生成消息,切到消费者类的控制台方法如下:

选择Cosumer,会看到如下信息:

由信息可以看到,消费者收到了同等数量的信息。

至此我们的第一个程序成功完成。下面将MQspring结合使用。

四、springactiveMQ的简单整合

在做上面的测试例子的时候我们可以看出发送一个简单的消息,要几行代码! 其实我们的目的就是发送和接受消息。幸运的是Spring为我们提供了大量的模板。spring为我们提供了JMSTemplate模板。

1 、引入spring相关文件和配置

我使用的是spring版本是3.1.0.M2,其他版本的也可以,只是配置不同,去spring官网下载zip包,解开后将dist目录下的所有jar包(根据自己选择)拷贝到项目lib目录下并加入项目library中,具体可以参考上面步骤。

src目录下新建applicationContext.xml文件并输入一下内容:

<? xml version =“1.0” encoding =“GBK” ?>

<beans xmlns =“http://www.springframework.org/schema/beans”

xmlns:xsi =“http://www.w3.org/2001/XMLSchema-instance”

xmlns:aop =“http://www.springframework.org/schema/aop”

xmlns:tx =“http://www.springframework.org/schema/tx”

xmlns:context =“http://www.springframework.org/schema/context”

xsi:schemaLocation =

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

http://www.springframework.org/schema/tx

http://www.springframework.org/schema/tx/spring-tx-3.0.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context-3.0.xsd

http://www.springframework.org/schema/aop

http://www.springframework.org/schema/aop/spring-aop-3.0.xsd”

default-autowire =“byName” >

</ beans>

打开web.xml并将其内容修改为以下内容:

<? xml version =“1.0” encoding =“GBK” ?>

<web-app version =“2.5” xmlns =“http://java.sun.com/xml/ns/javaee”

xmlns:xsi =“http://www.w3.org/2001/XMLSchema-instance”

xsi:schemaLocation =“http://java.sun.com/xml/ns/javaee

http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd” >

<context-param >

<param-name >contextConfigLocation </param-name >

<param-value >classpath*:applicationContext*.xml </param-value >

</ context-param>

<servlet >

<servlet-name >spring </servlet-name >

<servlet-class >

org.springframework.web.servlet.DispatcherServlet

</ servlet-class>

<load-on-startup >1</ load-on-startup>

</ servlet>

<servlet-mapping >

<servlet-name >spring </servlet-name >

<url-pattern >/</ url-pattern>

</ servlet-mapping>

<welcome-file-list >

<welcome-file >index.jsp </welcome-file >

</ welcome-file-list>

</ web-app>

2 、配置JMSTemplate模板

类似于jdbcTemplate,首先要配置一个ConnectionFactory,之后要开始配置JmsTemplate模板了。最后是配置消息目标了。消息分为队列和主题两大类。在applicationContext.xml中加入如下内容:

<!– 配置Jms模板 –>

<bean id =“jmsTemplate”

class =“org.springframework.jms.core.JmsTemplate” >

<property name =“connectionFactory” ref =“connectionFactory” />

<property name =“defaultDestination” ref =“destination” />

<property name =“receiveTimeout” value =“10000” />

</ bean>

<!– 配置JMS连接工厂 –>

<bean id =“connectionFactory”

class =“org.apache.activemq.ActiveMQConnectionFactory” >

<property name =“brokerURL” value =“tcp://localhost:61616″ />

</ bean>

<!– 发送消息的目的地(队列) –>

<bean id =“destination”

class =“org.apache.activemq.command.ActiveMQQueue” >

<!– 设置消息队列的名字 –>

<constructor-arg index =“0” value =“myQueue” />

</ bean>

配置完成了,那么如何使用JmsTemplate发送消息呢?

springbeanfactory得到一个jmsTemplate的实例和消息目标的实例,发送消息,够简单的吧。下面我们就来编写具体代码。

3 、编写代码

新建类ProducerService.java,界面如下:

代码如下:

package com.mq.service;

public interface ProducerService {

public void send();

}

新建一个生产者实现类ProducerServiceImpl .java,界面略,代码如下:

package com.mq.service;

import java.util.Date;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import javax.jms.Message;

import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

public class ProducerServiceImpl implements ProducerService {

JmsTemplate jmsTemplate ;

Destination destination ;

public void send() {

MessageCreator messageCreator = new MessageCreator(){

public Message createMessage(Session session) throws JMSException {

MapMessage message = session.createMapMessage();

message.setLong( “count”, new Date().getTime());

System. out.println( “–发送消息:+ new Date());

return message;

}

};

jmsTemplate .send( this.destination ,messageCreator);

}

public void setJmsTemplate(JmsTemplate jmsTemplate) {

this.jmsTemplate = jmsTemplate;

}

public void setDestination(Destination destination) {

this.destination = destination;

}

}

生产者编写完了,下面我们来编写消费者,上面说了,发送消息的时候, springbeanfactory得到一个jmsTemplate的实例和消息目标的实例,然后发送,那么接受的时候肯定也是得到一个jmsTemplate的实例和消息目标的实例,然后接受,下面我们来看具体代码。

新建一个消费者的借口和实现类,具体代码如下:

ConsumerService .java

package com.mq.service;

public interface ConsumerService {

public void receive();

}

ConsumerServiceImpl .java

package com.mq.service;

import java.util.Date;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.MapMessage;

import org.springframework.jms.core.JmsTemplate;

public class ConsumerServiceImpl implements ConsumerService {

JmsTemplate jmsTemplate ;

Destination destination ;

public void receive() {

MapMessage message = (MapMessage)jmsTemplate .receive();

try {

System. out.println( “–收到消息:+ new Date(message.getLong(“count” )));

} catch (JMSException e) {

e.printStackTrace();

}

}

public void setJmsTemplate(JmsTemplate jmsTemplate) {

this.jmsTemplate = jmsTemplate;

}

public void setDestination(Destination destination) {

this.destination = destination;

}

}

代码编写完毕,下面要进行bean的配置,在applicationContext.xml中加入如下代码:

<bean id =“producerService”

class =“com.mq.service.ProducerServiceImpl” >

<property name =“jmsTemplate” ref =“jmsTemplate” />

<property name =“destination” ref =“destination” />

</ bean>

<bean id =“consumerService”

class =“com.mq.service.ConsumerServiceImpl” >

<property name =“jmsTemplate” ref =“jmsTemplate” />

<property name =“destination” ref =“destination” />

</ bean>

附上完整的applicationContext.xml的内容:

<? xml version =“1.0” encoding =“GBK” ?>

<beans xmlns =“http://www.springframework.org/schema/beans”

xmlns:xsi =“http://www.w3.org/2001/XMLSchema-instance”

xmlns:aop =“http://www.springframework.org/schema/aop”

xmlns:tx =“http://www.springframework.org/schema/tx”

xmlns:context =“http://www.springframework.org/schema/context”

xsi:schemaLocation =

http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans-3.0.xsd

http://www.springframework.org/schema/tx

http://www.springframework.org/schema/tx/spring-tx-3.0.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context-3.0.xsd

http://www.springframework.org/schema/aop

http://www.springframework.org/schema/aop/spring-aop-3.0.xsd”

default-autowire =“byName” >

<bean id =“producerService”

class =“com.mq.service.ProducerServiceImpl” >

<property name =“jmsTemplate” ref =“jmsTemplate” />

<property name =“destination” ref =“destination” />

</ bean>

<bean id =“consumerService”

class =“com.mq.service.ConsumerServiceImpl” >

<property name =“jmsTemplate” ref =“jmsTemplate” />

<property name =“destination” ref =“destination” />

</ bean>

<!– 配置Jms模板 –>

<bean id =“jmsTemplate”

class =“org.springframework.jms.core.JmsTemplate” >

<property name =“connectionFactory” ref =“connectionFactory” />

<property name =“defaultDestination” ref =“destination” />

<property name =“receiveTimeout” value =“10000” />

</ bean>

<!– 配置JMS连接工厂 –>

<bean id =“connectionFactory”

class =“org.apache.activemq.ActiveMQConnectionFactory” >

<property name =“brokerURL” value =“tcp://localhost:61616″ />

</ bean>

<!– 发送消息的目的地(队列) –>

<bean id =“destination”

class =“org.apache.activemq.command.ActiveMQQueue” >

<!– 设置消息队列的名字 –>

<constructor-arg index =“0” value =“myQueue” />

</ bean>

</ beans>

4 、编写测试程序

需要的业务代码都已编写完毕,下面编写测试代码。新建一个生产者的测试类 ProducerTest.java。具体代码如下:

package com.mq.service;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ProducerTest {

private static ApplicationContext appContext = new ClassPathXmlApplicationContext(

“applicationContext.xml” );

private static void send() {

ProducerService producerService = (ProducerService) appContext

.getBean( “producerService”);

producerService.send();

}

/**

* @param args

*/

public static void main(String[] args) {

send();

}

}

再建一个消费者的测试类,ConsumerTest .java,具体代码如下:

package com.mq.service;

import org.springframework.context.ApplicationContext;

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class ConsumerTest {

private static ApplicationContext appContext = new ClassPathXmlApplicationContext(

“applicationContext.xml” );

private static void receive() {

ConsumerService consumerService = (ConsumerService) appContext

.getBean( “consumerService”);

consumerService.receive();

}

/**

* @param args

*/

public static void main(String[] args) {

receive();

}

}

5 、运行程序

所有代码都编写完了,我们来看一下我们的劳动成果。运行生产者测试类。控制台打印出如下内容,红线标注的冒号后面就是我们发送的内容:

运行消费者测试类。控制台打印出如下内容,红线标注的冒号后面就是我们接受的内容:

两者是一样的。

看来我们的劳动成果没有白费。以上是一些AciveMQ的基本应用,在接下来的时间,我会写一些高级应方面的文章,敬请关注本人百度空间和新郎微博。也欢迎各位通过一下两种途径和我进行交流,共同进步。推荐一下本人的微博,作为程序员,生活是非常枯燥的(一般都是很枯燥的,哪些以此为乐的不涉及),平时要自己丰富一下自己的业余生活,看些搞笑微博之类的。我每天都会发布一些糗事,欢迎大家关注。另外我在网上看到过有句话–程序员活该你单身身的 ,程序员招谁惹谁了啊,我们就应该单身。不过话又说回来了,程序员的生活确实单调,感情不丰富,不会说话,但是不能怨我们,其实我们也不想这样。推荐关注新浪微博一千零一夜话,微博地址: http://weibo.com/1968007227/,每天发布一条夜话,你懂得,希望各位程序猿每天能给自己的另一位一句温馨的话,不要成为苦逼的娃。

五、参考内容

1http://yunzhongxia.iteye.com/blog/566727

2ActiveMQ in action

3apache官网



快乐成长 每天进步一点点