ActiveMQ异步发送JMS消息

 在下载activeMq-all-5.2.jar.开发;

本实例来自apache 的官方网站:

package cn.com.vnvtrip.apachemq;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 此代码来自官方网站： 自己添加注释：
 *  总结：使用JMS的步骤：
 *  （1）创建一個JMS连接工厂
 *  （2）创建 一個JMS连接并启动连接
 * （3）创建一個JMS的会话并建立相应的JMS目标對象 (队列)
 * （4）通过相应的消息创建者创建消息和消息消费者接受消息。
 *
 * Hello world!
 */
public class App {

 public static void main(String[] args) throws Exception {
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  Thread.sleep(1000);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  Thread.sleep(1000);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldProducer(), false);
  Thread.sleep(1000);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldConsumer(), false);
  thread(new HelloWorldProducer(), false);
 }

 public static void thread(Runnable runnable, boolean daemon) {
  Thread brokerThread = new Thread(runnable);
  brokerThread.setDaemon(daemon);
  brokerThread.start();
 }

 /**
  * 消息的创建者的应用
  *
  * @author longgangbai
  *
  */
 public static class HelloWorldProducer implements Runnable {
  public void run() {
   try {
    // 创建一個ActionMQ的连接工厂
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
      "vm://localhost");

    // 创建一個ActionMQ连接工厂
    Connection connection = connectionFactory.createConnection();
    connection.start();

    // 创建一個ActionMQ会话
    Session session = connection.createSession(false,
      Session.AUTO_ACKNOWLEDGE);

    // 创建一個连接ActionMQ的目标(Topic or Queue)
    Destination destination = session.createQueue("TEST.FOO");

    // 创建一個消息的创建者从会话中队列目标中
    MessageProducer producer = session.createProducer(destination);
    producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

    // 创建一個消息
    String text = "Hello world! From: "
      + Thread.currentThread().getName() + " : "
      + this.hashCode();
    TextMessage message = session.createTextMessage(text);

    // 发送一個消息
    System.out.println("Sent message: " + message.hashCode()
      + " : " + Thread.currentThread().getName());
    producer.send(message);
    // 关闭会话和连接
    session.close();
    connection.close();
   } catch (Exception e) {
    System.out.println("Caught: " + e);
    e.printStackTrace();
   }
  }
 }

 /**
  * 消息的消费者
  *
  * @author longgangbai
  *
  */
 public static class HelloWorldConsumer implements Runnable,
   ExceptionListener {
  public void run() {
   try {

    // 创建一個JMS的连接工厂的
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
      "vm://localhost");

    // 创建一個JMS的连接
    Connection connection = connectionFactory.createConnection();
    // 启动连接對象
    connection.start();

    connection.setExceptionListener(this);

    // 创建会话
    Session session = connection.createSession(false,
      Session.AUTO_ACKNOWLEDGE);

    // 创建目标的對象 Create the destination (Topic or Queue)
    Destination destination = session.createQueue("TEST.FOO");

    // 创建消息的消费者的 Create a MessageConsumer from the Session to the
    // Topic or
    // Queue
    MessageConsumer consumer = session.createConsumer(destination);

    // 接受相应的消息對應 Wait for a message
    Message message = consumer.receive(1000);

    if (message instanceof TextMessage) {
     TextMessage textMessage = (TextMessage) message;
     String text = textMessage.getText();
     System.out.println("Received: " + text);
    } else {
     System.out.println("Received: " + message);
    }
    // 关闭会话和连接
    consumer.close();
    session.close();
    connection.close();
   } catch (Exception e) {
    System.out.println("Caught: " + e);
    e.printStackTrace();
   }
  }

  /*
   * 异步消息异常处理 (non-Javadoc)
   *
   * @see javax.jms.ExceptionListener#onException(javax.jms.JMSException)
   */
  public synchronized void onException(JMSException ex) {
   System.out.println("JMS Exception occured.  Shutting down client.");
  }
 }
}