你的位置:首页 > Java教程

[Java教程]ActiveMQ基本配置与示例演示

一、下载ActiveMQ

去官方网站下载:http://activemq.apache.org/

二、运行ActiveMQ

将apache-activemq-5.11.1-bin.zip解压,由于本系统是32位,所以进入apache-activemq-5.11.1\bin\win32目录。

1、安装InstallService.bat,如果出现下图,也许是你电脑已经安装过该服务。

image

2、点击wrapper.exe,运行,如果出现下图所示,表示已经简单配置好了。

image

不过期间,如果你没有设置JAVA_HOME环境变量,会出错,所以,最好是建立一个JAVA_HOME环境变量,其值设置为C:\Program Files\Java\jdk1.8.0(该路径为jdk的安装路径)。

3、要想运行activemq,必须开启activemq.bat服务,点击它,让其运行。

三、使用NetBeans创建项目运行

1、JMS基本概念

JMS(Java Message Service) 即Java消息服务。它提供标准的产生、发送、接收消息的接口简化企业应用的开发。它支持两种消息通信模型:点到点(point-to-point)(P2P)模型和发布/订阅(Pub/Sub)模型。P2P模型规定了一个消息只能有一个接收者;Pub/Sub 模型允许一个消息可以有多个接收者。
    对于点到点模型,消息生产者产生一个消息后,把这个消息发送到一个Queue(队列)中,然后消息接收者再从这个Queue中读取,一旦这个消息被一个接收者读取之后,它就在这个Queue中消失了,所以一个消息只能被一个接收者消费。

  与点到点模型不同,发布/订阅模型中,消息生产者产生一个消息后,把这个消息发送到一个Topic中,这个Topic可以同时有多个接收者在监听,当一个消息到达这个Topic之后,所有消息接收者都会收到这个消息。

2、编程的结构

2.1消息产生者向JMS发送消息的步骤 


(1)创建连接使用的工厂类JMS ConnectionFactory 
(2)使用管理对象JMS ConnectionFactory建立连接Connection 
(3)使用连接Connection 建立会话Session 
(4)使用会话Session和管理对象Destination创建消息生产者MessageSender 
(5)使用消息生产者MessageProducer发送消息


2.2消息消费者从JMS接受消息的步骤 


(1)创建连接使用的工厂类JMS ConnectionFactory 
(2)使用管理对象JMS ConnectionFactory建立连接Connection 
(3)使用连接Connection 建立会话Session 
(4)使用会话Session和管理对象Destination创建消息消费者MessageReceiver 
(5)使用消息消费者MessageConsumer接受消息

3、代码

整个工程需引入activemq-all-5.11.1.jar作为类库文件。其次建立JmsSender.java与JmsReceiver.java两个文件模拟消息产生者向JMS发送消息和消息消费者从JMS接受消息。

JmsSender.java文件

/* * To change this license header, choose License Headers in Project Properties. * To change this template file, choose Tools | Templates * and open the template in the editor. */package testactivemq;import javax.jms.*;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/** * * @author LIN NP */public class JmsSender{  private ConnectionFactory connectionFactory = null;  private Connection connection = null;  private Session session = null;  private Destination destination = null;  private MessageProducer producer = null;  private static final int SEND_NUMBER = 5;    /**   *   */  public void init()  {    // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar    connectionFactory = new ActiveMQConnectionFactory(        ActiveMQConnection.DEFAULT_USER,        ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616");  // ActiveMQ默认使用的TCP连接端口是61616        try{      // 构造从工厂得到连接对象      connection = connectionFactory.createConnection();      connection.start();            // 获取操作连接      session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);            /**      * 第一种方式:Queue      *///      destination = session.createQueue("xkey");    // "xkey"可以取其他的。//      producer = session.createProducer(destination); // 得到消息生成者【发送者】            /**       * 第二种方式:Topic       */             Topic topic = session.createTopic("xkey.Topic");       producer = session.createProducer(topic);             /**       *        */      // 设置不持久化,此处学习,实际根据项目决定      producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);            // 构造消息,此处写死,项目就是参数,或者方法获取      sendMessage(session,producer);      session.commit();                }    catch(Exception e)    {      e.printStackTrace();    }    finally    {      try       {        connection.close();      }       catch (JMSException e)       {        // TODO Auto-generated catch block        e.printStackTrace();      }    }  }    private void sendMessage(Session session,MessageProducer producer) throws JMSException  {    for (int i = 1; i <= SEND_NUMBER; i ++)     {       TextMessage message = session.createTextMessage("ActiveMq 发送的消息: " + i);       // 发送消息      System.out.println("发送消息:" + "ActiveMq 发送的消息: " + i);       producer.send(message);     }   }  /**   * @param args   */  public static void main(String[] args)   {    // TODO Auto-generated method stub    JmsSender jms = new JmsSender();    jms.init();  }}

JmsReceiver.java文件

/* * To change this license header, choose License Headers in Project Properties. * To change this template file, choose Tools | Templates * and open the template in the editor. */package testactivemq;import javax.jms.*;import org.apache.activemq.ActiveMQConnection;import org.apache.activemq.ActiveMQConnectionFactory;/** * * @author LIN NP */public class JmsReceiver{  private ConnectionFactory connectionFactory = null;  private Connection connection = null;  private Session session = null;  private MessageConsumer consumer = null;  private Destination destination = null;    public void init()  {        connectionFactory = new ActiveMQConnectionFactory(        ActiveMQConnection.DEFAULT_USER,        ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616"); // ActiveMQ默认使用的TCP连接端口是61616    try    {      // 构造从工厂得到连接对象      connection = connectionFactory.createConnection();      connection.start();      // 获取操作连接      session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);       /**      * 第一种方式:Queue      *///      destination = session.createQueue("xkey");//      consumer = session.createConsumer(destination);            /**       * 第二种方式:Topic       */      Topic topic = session.createTopic("xkey.Topic");      consumer = session.createConsumer(topic);            /**       *        */      while (true)       {         //设置接收者接收消息的时间,为了便于测试,这里谁定为500s        TextMessage message = (TextMessage) consumer.receive(500000);         if (null != message)         {           System.out.println("Receiver " + message.getText());         }        else         {           break;         }       }     }    catch(Exception e)    {      e.printStackTrace();    }    finally    {      try       {        connection.close();      }       catch (JMSException e)       {        // TODO Auto-generated catch block        e.printStackTrace();      }    }  }  /**   * @param args   */  public static void main(String[] args)   {    // TODO Auto-generated method stub    JmsReceiver jms = new JmsReceiver();    jms.init();  }}

4、测试过程

在整个过程中,要保证activemq.bat服务是运行的。

4.1 运行JmsReceiver.java文件,testActiveMQ (run)控制台会出现

log4j:WARN No appenders could be found for logger (org.apache.activemq.transport.WireFormatNegotiator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

4.2 运行JmsSender.java文件,testActiveMQ (run) #2控制台会出现

log4j:WARN No appenders could be found for logger (org.apache.activemq.transport.WireFormatNegotiator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
发送消息:ActiveMq 发送的消息: 1
发送消息:ActiveMq 发送的消息: 2
发送消息:ActiveMq 发送的消息: 3
发送消息:ActiveMq 发送的消息: 4
发送消息:ActiveMq 发送的消息: 5

其中警告信息可以忽略,暂时也没去找警告来自哪里。

4.3 返回到testActiveMQ (run)控制台会出现

run:
log4j:WARN No appenders could be found for logger (org.apache.activemq.transport.WireFormatNegotiator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Receiver ActiveMq 发送的消息: 1
Receiver ActiveMq 发送的消息: 2
Receiver ActiveMq 发送的消息: 3
Receiver ActiveMq 发送的消息: 4
Receiver ActiveMq 发送的消息: 5

 

主要参考:

http://www.cnblogs.com/xwdreamer/archive/2012/02/21/2360818.html

http://www.open-open.com/lib/view/open1388994166156.html