ActiveMQ简单实现之一对一生产和消费

package com.sgor.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
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;
/**
 * 消息生产者类
 * @author xixiao
 *
 */
public class Producer {
	public static void main(String[] args){
		ConnectionFactory connectionFactory; //创建链接工厂
		Connection connection = null;//链接
		Session session;//创建会话
		Destination destination;//消息目的地 消息队列
		MessageProducer messageProducer;//消息生产者
		//实例化链接工厂  参数为 用户,密码,url
		connectionFactory = new ActiveMQConnectionFactory("xixiao", "xixiao", ActiveMQConnection.DEFAULT_BROKER_URL);
		try {
			connection=connectionFactory.createConnection();//通过链接工厂创建链接
			connection.start();//启动链接
			//创建会话 Session.AUTO_ACKNOWLEDGE。receive 或MessageListener.onMessage()成功返回的时候，自动确认收到消息。
			session =connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			//创建一个消息队列名称为hello ActiveMQ 消息队列中可包含需要发布消息 
			destination = session.createQueue("Hello ActiveMQ");
			//将创建的消息队列hello ActiveMQ交给消息发布者messageProdecer
			messageProducer=session.createProducer(destination);
			for (int i = 0; i <5 ; i++) {
				//生产5条消息
				TextMessage message=session.createTextMessage(i+"条消息");
				System.out.println(message.getText());
　　　　　　　　　　　　　　　　　　　　//发布消息
				messageProducer.send(message);
			}
			//提交事物
			session.commit();
		} catch (JMSException e) {
			e.printStackTrace();
		}finally{
			try {
				//关闭连接
				connection.close();
			} catch (JMSException e) {
				e.printStackTrace();
			}
		}
		
	}
}

package com.sgor.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Consumer {
	public static void main(String[] args) {
		ConnectionFactory connectionFactory; 
		Connection connection = null;
		Session session;
		Destination destination;
		MessageConsumer messageConsumer;
		connectionFactory = new ActiveMQConnectionFactory("xixiao", "xixiao", ActiveMQConnection.DEFAULT_BROKER_URL);
		try {
			//通过链接工厂创建链接
			connection  = connectionFactory.createConnection();
			connection.start();//启动链接
			//这里直接设为false  不需要事物
			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			destination = session.createQueue("Hello ActiveMQ"); //创建消息队列 用于接收发布的消息
			messageConsumer = session.createConsumer(destination);
			/**
			 * 监听生产者方式接受消息，生产者产生消息才开始接收 需要监听器 实现MessageListener (javax.jms.MessageListener包)
			 */
			messageConsumer.setMessageListener(new Listener());
		} catch (JMSException e) {
			e.printStackTrace();
		}
		
		
	}
}

package com.sgor.activemq;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class Listener implements MessageListener{

	@Override
	public void onMessage(Message arg0) {
		try {
			System.out.println("消息："+((TextMessage)arg0).getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
		
	}

}