最近由于公司項(xiàng)目需要，開始學(xué)習(xí)JMS，用的是ActiveMQ。由于這方面網(wǎng)上的例子不是很多，而且有的也不完整。于是經(jīng)過(guò)幾天的摸索學(xué)習(xí)，寫了一個(gè)簡(jiǎn)單的小例子，現(xiàn)在貼出來(lái)與大家分享。
ProducerTool.java用于發(fā)送消息：

package homework;

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 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;

 private String subject = "TOOL.DEFAULT";

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageProducer producer = null;

 // 初始化
 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);
 }

 // 發(fā)送消息
 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!");
 }

 // 關(guān)閉連接
 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();
 }
}

package homework;

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 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;

 private String subject = "TOOL.DEFAULT";

 private Destination destination = null;

 private Connection connection = null;

 private Session session = null;

 private MessageConsumer consumer = null;

 // 初始化
 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);
  
 }

 // 消費(fèi)消息
 public void consumeMessage() throws JMSException, Exception ...{
  initialize();
  connection.start();
  
  System.out.println("Consumer:->Begin listening...");
  // 開始監(jiān)聽
  consumer.setMessageListener(this);
  // Message message = consumer.receive();
 }

 // 關(guān)閉連接
 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();
 }

 // 消息處理函數(shù)
 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) ...{
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

如果想主動(dòng)的去接受消息，而不用消息監(jiān)聽的話，把consumer.setMessageListener(this)改為Message message = consumer.receive()，手動(dòng)去調(diào)用MessageConsumer的receive方法即可。

下面是測(cè)試類Test.java：

 

package homework;

import javax.jms.JMSException;

public class Test ...{

 /** *//**
  * @param args
  */
 public static void main(String[] args) throws JMSException, Exception ...{
  // TODO Auto-generated method stub
  ConsumerTool consumer = new ConsumerTool();
  ProducerTool producer = new ProducerTool();
  // 開始監(jiān)聽
  consumer.consumeMessage();
  
  // 延時(shí)500毫秒之后發(fā)送消息
  Thread.sleep(500);
  producer.produceMessage("Hello, world!");
  producer.close();
  
  // 延時(shí)500毫秒之后停止接受消息
  Thread.sleep(500);
  consumer.close();
 }
}

 

以上就是我學(xué)習(xí)ActiveMQ之后寫的一個(gè)簡(jiǎn)單的例子，希望對(duì)你有幫助，如果有什么錯(cuò)誤還請(qǐng)指正。 

Trackback: http://tb.blog.csdn.net/TrackBack.aspx?PostId=1536460