ActiveMQ学习笔记----ActiveMQ和JBossMQ性能对比测试代码

本文描述了对ActiveMQ进行性能测试的代码。性能测试用源代码共包含3个文件，分别是：
JMS消息发送类：ActiveMQProducer.java
JMS消息接收类：ActiveMQConsumer.java
JMS消息收发测试主类：ActiveMQTest.java
下面分别介绍这三个类。

1. JMS消息发送类 ActiveMQProducer.java 的源码如下：
/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */

import javax.jms.*;

import org.apache.activemq.*;

public class ActiveMQProducer {
 public final static int MAX_SEND_TIMES = 100;

 private String user = ActiveMQConnection.DEFAULT_USER;

 private String password = ActiveMQConnection.DEFAULT_PASSWORD;

 private String url = ActiveMQConnection.DEFAULT_BROKER_URL;

 private String subject = "TOOL.DEFAULT";

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageProducer producer = null;

 /**
  * 初始化
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 private void initialize() throws JMSException, Exception {
  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    user, password, url);
  connection = connectionFactory.createConnection();
  session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  destination = session.createQueue(subject);
  producer = session.createProducer(destination);
  producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  connection.start();
 }

 /**
  * 发送消息
  * 
  * @param message
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public void produceMessage(String message) throws JMSException, Exception {
  initialize();
  TextMessage msg = session.createTextMessage(message);
  
  long beginTime = System.currentTimeMillis();

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

  for (int i = 0; i < MAX_SEND_TIMES; i++) {
   producer.send(msg);
   if ((i + 1) % 1000 == 0) {
    System.out.println("This is the " + i + " message!");
   }
  }

  System.out.println("Producer:->Message sent complete!");
  long endTime = System.currentTimeMillis();
  long executeTime = endTime - beginTime;
  System.out.println("ActiveMQ send" + MAX_SEND_TIMES + " messages used: "
    + executeTime + " ms");
 }

 /**
  * 关闭连接
  * 
  * @throws JMSException
  */
 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();
  }
 }
}

2. JMS消息接收类 ActiveMQConsumer.java 的源码如下
/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */

import javax.jms.*;

import org.apache.activemq.*;

public class ActiveMQConsumer implements MessageListener {
 public static int RECEIVED_MSG_NUM = 0;
 long beginReceiveTime = 0;
 long endReceiveTime = 0;
 long receiveDuringTime = 0;

 private String user = ActiveMQConnection.DEFAULT_USER;

 private String password = ActiveMQConnection.DEFAULT_PASSWORD;

 private String url = ActiveMQConnection.DEFAULT_BROKER_URL;

 private String subject = "TOOL.DEFAULT";

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageConsumer consumer = null;

 /**
  * 初始化
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 private void initialize() throws JMSException, Exception {
  ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
    user, password, url);
  connection = connectionFactory.createConnection();
  session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  destination = session.createQueue(subject);
  consumer = session.createConsumer(destination);
 }

 /**
  * 接收消息
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public void consumeMessage() throws JMSException, Exception {
  initialize();
  connection.start();
  System.out.println("Consumer:->Begin listening...");
  // 设置消息监听
  consumer.setMessageListener(this);
 }

 /**
  * 关闭连接
  * 
  * @throws JMSException
  */
 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();
  }
 }

 /**
  * 收到消息的处理
  * 
  * @param message
  */
 public void onMessage(Message message) {
  try {
   if (message instanceof TextMessage) {

    TextMessage txtMsg = (TextMessage) message;
    String msg = txtMsg.getText();

    // receive the first message
    if (RECEIVED_MSG_NUM == 0) {
     beginReceiveTime = System.currentTimeMillis();
    }

    RECEIVED_MSG_NUM++;

    // print one String when received 1000 message
    if ((RECEIVED_MSG_NUM + 1) % 1000 == 0) {
     System.out.println("Consumer:->Received: "
       + RECEIVED_MSG_NUM);
    }

    // receive the last message
    if (RECEIVED_MSG_NUM == ActiveMQProducer.MAX_SEND_TIMES - 1) {
     endReceiveTime = System.currentTimeMillis();
     receiveDuringTime = endReceiveTime - beginReceiveTime;
     System.out.println("ActiveMQ Receive "
       + ActiveMQProducer.MAX_SEND_TIMES
       + " messages used: " + receiveDuringTime + " ms");
    }

   } else {
    System.out.println(System.currentTimeMillis()
      + "Consumer:->Received: " + message);
   }
  } catch (JMSException e) {
   e.printStackTrace();
  }
 }
}

3. JMS消息收发测试主类：ActiveMQTest.java
/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */
import javax.jms.*;

public class ActiveMQTest {

 /**
  * 
  * @param args
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public static void main(String[] args) throws JMSException, Exception {
  ActiveMQConsumer consumer = new ActiveMQConsumer();
  ActiveMQProducer producer = new ActiveMQProducer();
  char[] tempChars=new char[1024];
  for(int i=0;i<1024;i++){
   tempChars[i]='a';
  }
  String tempMsg=String.valueOf(tempChars);
  // 开始监听
  consumer.consumeMessage();
  producer.produceMessage(tempMsg);
  producer.close();

  // 延时5000ms后关闭连接
  Thread.sleep(5000);
  consumer.close();
 }
}

为了与JBossMQ进行性能对比测试，下面把JBossMQ收发消息的测试代码也一并附上。
为了保证测试代码的一致性，JBossMQ性能测试的代码也包含3个文件，分别是：
JMS消息发送类：JBossMQProducer.java
JMS消息接收类：JBossMQConsumer.java
JMS消息收发测试主类：JBossMQTest.java
下面分别介绍这三个类。

1. JMS消息发送类 JBossMQProducer.java 的源码如下：
/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */

import java.util.Properties;

import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;

public class JbossMQProducer {
 public final static int MAX_SEND_TIMES = 100;

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageProducer producer = null;

 /**
  * 初始化
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 private void initialize() throws JMSException, Exception {
  Properties props = new Properties();
  props.put(Context.INITIAL_CONTEXT_FACTORY,
    "org.jnp.interfaces.NamingContextFactory");
  props.put(Context.URL_PKG_PREFIXES, "org.jboss.naming");
  props.put(Context.PROVIDER_URL, "localhost:1099");

  InitialContext jmsContext = new InitialContext(props);
  ConnectionFactory connectionFactory = (ConnectionFactory) jmsContext
    .lookup("ConnectionFactory");
  connection = connectionFactory.createConnection();
  session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  destination = (Queue) jmsContext.lookup("queue/A");
  producer = session.createProducer(destination);
  producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  connection.start();
 }

 /**
  * 发送消息
  * 
  * @param message
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public void produceMessage(String message) throws JMSException, Exception {
  initialize();
  TextMessage msg = session.createTextMessage(message);

  long beginTime = System.currentTimeMillis();

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

  for (int i = 0; i < MAX_SEND_TIMES; i++) {
   if ((i + 1) % 1000 == 0) {
    System.out.println("This is the " + i + " message!");
   }
   producer.send(msg);
  }

  System.out.println("Producer:->Message sent complete!");
  long endTime = System.currentTimeMillis();
  long executeTime = endTime - beginTime;
  System.out.println("JbossMQ send" + MAX_SEND_TIMES + " messages used: "
    + executeTime + " ms");
 }

 /**
  * 关闭连接
  * 
  * @throws JMSException
  */
 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();
  }
 }
}

2. JMS消息接收类 JBossMQConsumer.java 的源码如下
/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */

import java.util.Properties;

import javax.jms.*;
import javax.naming.*;

public class JbossMQConsumer implements MessageListener {
 public static int RECEIVED_MSG_NUM = 0;
 long beginReceiveTime = 0;
 long endReceiveTime = 0;
 long receiveDuringTime = 0;

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageConsumer consumer = null;

 /**
  * 初始化
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 private void initialize() throws JMSException, Exception {
  Properties props = new Properties();
  props.put(Context.INITIAL_CONTEXT_FACTORY,
    "org.jnp.interfaces.NamingContextFactory");
  props.put(Context.URL_PKG_PREFIXES, "org.jboss.naming");
  props.put(Context.PROVIDER_URL, "localhost:1099");

  // get naming context
  InitialContext jmsContext = new InitialContext(props);
  ConnectionFactory connectionFactory = (ConnectionFactory) jmsContext
    .lookup("ConnectionFactory");

  connection = connectionFactory.createConnection();
  session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  destination = (Queue) jmsContext.lookup("queue/A");
  consumer = session.createConsumer(destination);
 }

 /**
  * 接收消息
  * 
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public void consumeMessage() throws JMSException, Exception {
  initialize();
  connection.start();
  System.out.println("Consumer:->Begin listening...");
  // 设置监听
  consumer.setMessageListener(this);
 }

 /**
  * 关闭连接
  * 
  * @throws JMSException
  */
 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();
  }
 }

 /**
  * 处理消息
  * 
  * @param message
  */
 public void onMessage(Message message) {
  try {
   if (message instanceof TextMessage) {

    TextMessage txtMsg = (TextMessage) message;
    String msg = txtMsg.getText();

    // receive the first message
    if (RECEIVED_MSG_NUM == 0) {
     beginReceiveTime = System.currentTimeMillis();
    }

    RECEIVED_MSG_NUM++;

    //print one String when received 1000 message
    if ((RECEIVED_MSG_NUM + 1) % 1000 == 0) {
     System.out.println("Consumer:->Received: "
       + RECEIVED_MSG_NUM);
    }

    // receive the last message
    if (RECEIVED_MSG_NUM == JbossMQProducer.MAX_SEND_TIMES - 1) {
     endReceiveTime = System.currentTimeMillis();
     receiveDuringTime = endReceiveTime - beginReceiveTime;
     System.out.println("JbossMQ receive "
       + JbossMQProducer.MAX_SEND_TIMES
       + " messages used: " + receiveDuringTime + " ms");
    }
        
   } else {
    System.out.println(System.currentTimeMillis()
      + "Consumer:->Received: " + message);
   }
  } catch (JMSException e) {
   e.printStackTrace();
  }
 }
}

3. JMS消息收发测试主类：JBossMQTest.java
import javax.jms.JMSException;

/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2007</p>
 * <p>Company: </p>
 * @author mqboss
 * @version 1.0
 */

public class JbossMQTest {

 /**
  * 
  * @param args
  * @throws JMSException
  * @throws java.lang.Exception
  */
 public static void main(String[] args) throws JMSException, Exception {
  JbossMQConsumer consumer = new JbossMQConsumer();
  JbossMQProducer producer = new JbossMQProducer();
  char[] tempChars = new char[1024];
  for (int i = 0; i < 1024; i++) {
   tempChars[i] = 'a';
  }
  String tempMsg = String.valueOf(tempChars);
  // 启动消息监听
  consumer.consumeMessage();
  producer.produceMessage(tempMsg);
  producer.close();
  // 5000ms后关闭连接
  Thread.sleep(5000);
  consumer.close();
 }
}

以上是把ActiveMQ集成到JBoss以后，对ActiveMQ和JBossMQ进行性能对比测试的源代码。
很显然，以上源代码基本一致，所以性能测试结果可以很好的说明ActiveMQ和JBossMQ的性能对比情况。

参考文献：
[1]ActiveMQ的一个简单示例