jms和ActiveMQ的一个小例子

package com.example;


import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.jms.TopicPublisher;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;

public class My implements MessageListener{

//	队列的连接工厂和连接
	private QueueConnectionFactory queueConnectionFactory;
	private QueueConnection queueConnection;
//	主题的连接工厂和连接
	private TopicConnectionFactory topicConnectinFactory;
	private TopicConnection topicConnection;
//	主题持久订阅的设置
	private String clientId;
	private String durableName;
//	是否为主题的持久订阅
	private boolean isDurable = false;
//	主题会话
	private TopicSession topicSession;
//	主题发布者
	private TopicPublisher  topicPublisher;
//	主题订阅者
	private TopicSubscriber topicSubscriber;
//	发布的主题
	private Topic topicPublish;
//	发布的主题名称
	private String topicPublishName;
//	订阅的主题名称
	private String topicSubscribName;
//	订阅的主题
	private Topic topicSubscrib;
//	是否采用主题模式
	private boolean isTopic = false;
//	队列会话
	private QueueSession queueSession;
//	队列发送者
	private QueueSender queueSender;
//	队列接收者
	private QueueReceiver queueReceiver;
//	发送的队列
	private Queue queueSend;
//  接收的队列
	private Queue queueReceive;
//	发送的队列名称
	private String queueSendName;
//	接收的队列名称
	private String queueReceiveName;
//	是否采用事务
	private boolean isTransacted = false;
//	应答的参数
	private int acknowledgementMode;
//	连接工厂的参数
	private String user = ActiveMQConnection.DEFAULT_USER;
	private String password = ActiveMQConnection.DEFAULT_PASSWORD;
	private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
//	是否在发送消息用持久模式
	private boolean isPersistent = false;
//	消息的过期时间
	private long timeToLive;
	
	
	public QueueConnectionFactory getQueueConnectionFactory() {
		if(this.queueConnectionFactory==null)
			this.queueConnectionFactory = new ActiveMQConnectionFactory(this.getUser(),this.getPassword(),this.getUrl());
		return queueConnectionFactory;
	}

	public QueueConnection getQueueConnection() {
		if(this.queueConnection==null)
		{

			try {
				this.queueConnection = this.getQueueConnectionFactory().createQueueConnection();
				this.queueConnection.start();
				
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return queueConnection;
	}

	public TopicConnectionFactory getTopicConnectinFactory() {
		if(this.topicConnectinFactory==null)
		{
			this.topicConnectinFactory = new ActiveMQConnectionFactory(this.getUser(),this.getPassword(),this.getUrl());
		}
		return topicConnectinFactory;
	}

	public TopicConnection getTopicConnection() {
		if(this.topicConnection==null)
		{
			try {
				this.topicConnection = this.getTopicConnectinFactory().createTopicConnection();
				
				if(this.isDurable && this.getClientId()!=null && this.getClientId().length()>0 && !"null".equals(this.getClientId()))
				{
					this.topicConnection.setClientID(this.getClientId());
				}
				this.topicConnection.start();
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return topicConnection;
	}

	public String getClientId() {
		return clientId;
	}
	public void setClientId(String clientId) {
		this.clientId = clientId;
	}
	public String getDurableName() {
		return durableName;
	}
	public void setDurableName(String durableName) {
		this.durableName = durableName;
	}
	public boolean isDurable() {
		return isDurable;
	}
	public void setDurable(boolean isDurable) {
		this.isDurable = isDurable;
	}
	public TopicSession getTopicSession() {
		if(this.topicSession==null)
		{
			try {
			this.topicSession = this.getTopicConnection().createTopicSession(this.isTransacted(), this.getAcknowledgementMode());
			} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			}
		}
		return topicSession;
	}

	public TopicPublisher getTopicPublisher() {
		if(this.topicPublisher == null)
		{
			try {
				this.topicPublisher = this.getTopicSession().createPublisher(this.getTopicPublish());
//				是否采用发送方的持久及消息过期的设置
				if(this.isPersistent)
				{
					this.topicPublisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
					this.topicPublisher.setTimeToLive(this.getTimeToLive());
				}
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return topicPublisher;
	}

	public TopicSubscriber getTopicSubscriber() {
		if(this.topicSubscriber==null)
		{
			try {
				if(this.isDurable && this.isTopic )
				{
					this.topicSubscriber = this.getTopicSession().createDurableSubscriber(this.getTopicSubscrib(), this.getDurableName());
				}
				else
				{
					this.topicSubscriber = this.getTopicSession().createSubscriber(this.getTopicSubscrib());
					
				}
//				采用监听的方式接收消息，也可以采用主动接收的方式
				this.topicSubscriber.setMessageListener(this);
				this.getTopicConnection().start();
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return topicSubscriber;
	}

	public Topic getTopicPublish() {
		this.topicPublish = new ActiveMQTopic(this.getTopicPublishName());
		return topicPublish;
	}

	public String getTopicPublishName() {
		return topicPublishName;
	}
	public void setTopicPublishName(String topicPublishName) {
		this.topicPublishName = topicPublishName;
	}
	public String getTopicSubscribName()
	{
		return topicSubscribName;
	}
	public void setTopicSubscribName(String topicSubscribName)
	{
		this.topicSubscribName = topicSubscribName;
	}
	public Topic getTopicSubscrib() {
		this.topicSubscrib = new ActiveMQTopic(this.getTopicSubscribName());
		return topicSubscrib;
	}

	public boolean isTopic() {
		return isTopic;
	}
	public void setTopic(boolean isTopic) {
		this.isTopic = isTopic;
	}
	public QueueSession getQueueSession() {
		if(this.queueSession==null)
		{
			try {
				this.queueSession = this.getQueueConnection().createQueueSession(this.isDurable(), this.getAcknowledgementMode());
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return queueSession;
	}

	public QueueSender getQueueSender() {
		if(this.queueSender==null)
		{
			try {
				this.queueSender = this.getQueueSession().createSender(this.getQueueSend()) ;
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		return queueSender;
	}

	public QueueReceiver getQueueReceiver()throws JMSException {
		if(queueReceiver==null)
		{
			this.queueReceiver = this.getQueueSession().createReceiver(this.getQueueReceive());
//			采用监听的方式接收消息，也可以采用主动接收的方式
			this.queueReceiver.setMessageListener(this);
//			this.queueConnection.start();
		}
		return queueReceiver;
	}

	public Queue getQueueSend() {
		this.queueSend = new ActiveMQQueue(this.getQueueSendName());
		return queueSend;
	}

	public 	Queue getQueueReceive()
	{
		this.queueReceive = new ActiveMQQueue(this.getQueueReceiveName());
		return queueReceive;
	}

	public String getQueueSendName() {
		return queueSendName;
	}
	public void setQueueSendName(String queueSendName) {
		this.queueSendName = queueSendName;
	}
	public String getQueueReceiveName() {
		return queueReceiveName;
	}
	public void setQueueReceiveName(String queueReceiveName) {
		this.queueReceiveName = queueReceiveName;
	}
	public boolean isTransacted() {
		return isTransacted;
	}
	public void setTransacted(boolean isTransacted) {
		this.isTransacted = isTransacted;
	}
	public int getAcknowledgementMode() {
		return acknowledgementMode;
	}
	public void setAcknowledgementMode(int acknowledgementMode) {
		this.acknowledgementMode = acknowledgementMode;
	}
	public String getUser() {
		return user;
	}
	public void setUser(String user) {
		this.user = user;
	}
	public String getPassword() {
		return password;
	}
	public void setPassword(String password) {
		this.password = password;
	}
	public String getUrl() {
		return url;
	}
	public void setUrl(String url) {
		this.url = url;
	}
	public boolean isPersistent() {
		return isPersistent;
	}
	public void setPersistent(boolean isPersistent) {
		this.isPersistent = isPersistent;
	}
	public long getTimeToLive() {
		return timeToLive;
	}
	public void setTimeToLive(long timeToLive) {
		this.timeToLive = timeToLive;
	}
//	发布主题消息--字符串型的消息，也可以发布其他的消息类型
	public void publishTextMessage(String textMessage) throws JMSException
	{
		TextMessage message = this.getTopicSession().createTextMessage();
		message.clearBody();
		message.setText(textMessage);
		this.getTopicPublisher().publish(message);
		
		if(this.isTransacted)
		{
			this.getTopicSession().commit();
		}
		
	}
//	发送队列消息--字符串型的消息，也可以发送其他的消息类型
	public void sendTextMessage(String textMessage) throws JMSException
	{
		TextMessage message = this.getQueueSession().createTextMessage();
		message.clearBody();
		message.setText(textMessage);
		this.getQueueSender().send(message);
		if(this.isTransacted)
		{
			this.getQueueSession().commit();
		}
	}
//	监听接口的方法，用于接收消息
	public void onMessage(Message message) {
		// TODO Auto-generated method stub
		if(message instanceof  TextMessage)
		{
			TextMessage myMessage = (TextMessage)message;
			try {
				
				System.out.println(myMessage.getText());
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		if(this.isTransacted)
		{
			try {
				if(this.isTopic)
				{
					this.getTopicSession().commit();
				}
				else
				{
					this.getQueueSession().commit();
				}
				
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
//	关闭相应的资源
	public void closeResource() throws JMSException
	{ 
		if(this.isTopic)
		{
			if(this.topicPublisher!=null)
			{
				this.topicPublisher.close();
			}
			if(this.topicSubscriber!=null)
			{
				this.topicSubscriber.close();
			}
			if(this.topicSession!=null)
			{
				this.topicSession.close();
			}
			if(this.topicConnection!=null)
			{
				this.topicConnection.close();
			}
			System.out.println("TopicResource is closed..");
		}
		else
		{
			if(this.queueSender!=null)
			{
				this.queueSender.close();
			}
			if(this.queueReceiver!=null)
			{
				this.queueReceiver.close();
			}
			if(this.queueSession!=null)
			{
				this.queueSession.close();
			}
			if(this.queueConnection!=null)
			{
				this.queueConnection.close();
			}
			System.out.println("QueueResource is closed..");
		}
		
	}
//  测试
	public static void main(String[] args) throws JMSException
	{

//		队列  接收端的测试
//		My receivequeue = new My();
//		receivequeue.setDurable(false);
//		receivequeue.setTopic(false);
//		receivequeue.setTransacted(false);
//		receivequeue.setPersistent(false);
//		receivequeue.setAcknowledgementMode(QueueSession.AUTO_ACKNOWLEDGE);
//		receivequeue.setQueueReceiveName("kkk");
//		receivequeue.getQueueReceiver();
//		synchronized(receivequeue)
//		{
//			try {
//				receivequeue.wait(900);
//			} catch (InterruptedException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
//		}
//		receivequeue.closeResource();
		

		
//		主题  接收端得测试
		My subscribtopic = new My();
		subscribtopic.setDurable(true);
		subscribtopic.setClientId("opop");
		subscribtopic.setDurableName("ww");
		subscribtopic.setPersistent(false);
		subscribtopic.setTopic(true);
		subscribtopic.setTransacted(false);
		subscribtopic.setAcknowledgementMode(TopicSession.AUTO_ACKNOWLEDGE);
		subscribtopic.setTopicSubscribName("eee");
		subscribtopic.getTopicSubscriber();
	    synchronized(subscribtopic)
	    {
	    	try {
				subscribtopic.wait(900);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	    }
	    subscribtopic.closeResource();
		
		
	}
	
	
}//end class

以上是队列和主题的编码，下面是启动ActiveMQ的类

package com.by.ptop;

import org.apache.activemq.broker.BrokerService;

public class CreateServer {

	public void startServer()
	{
		BrokerService broker = new BrokerService();
//		配置jms服务器
		try {
			broker.addConnector("tcp://localhost:61616");
			broker.start();
			System.out.println("jms服务器已经启动.");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public static void main(String[] args)
	{
		CreateServer cs = new CreateServer();
		cs.startServer();
	}
}

下面是测试用的：

package com.example;

import javax.jms.JMSException;
import javax.jms.QueueSession;
import javax.jms.TopicSession;

import com.by.ptop.CreateServer;

public class MyTest {

	
	public static void main(String[] args) throws JMSException {
//		启动jms服务器
		CreateServer cs = new CreateServer();
		cs.startServer();

		for(int i = 1;i<=9;i++)
		{
//			队列  发送端得编程
//			My sendqueue = new My();
//			sendqueue.setDurable(false);
//			sendqueue.setPersistent(false);
//			sendqueue.setTransacted(false);
//			sendqueue.setTopic(false);
//			sendqueue.setAcknowledgementMode(QueueSession.AUTO_ACKNOWLEDGE);
//			sendqueue.setQueueSendName("kkk");
//			sendqueue.sendTextMessage("zz"+i);
//			sendqueue.closeResource();
			
//			主题 接收端得编程
			My publishtopic = new My();
			publishtopic.setDurable(true);
			publishtopic.setClientId("opop");
			publishtopic.setDurableName("ww");
			publishtopic.setPersistent(false);
			publishtopic.setTransacted(false);
			publishtopic.setTopic(true);
			publishtopic.setAcknowledgementMode(TopicSession.AUTO_ACKNOWLEDGE);
			publishtopic.setTopicPublishName("eee");
			publishtopic.publishTextMessage("zz"+i);
			publishtopic.closeResource();
		}
		
		

	}

}

以上是本人初学jms时写的，如有不足之处或者您有更好的代码，还望各位不吝赐教。小弟在这里先谢谢大家了。