JMS之ActiveMQ--点对点消息实现

痛定思痛。 2022-05-20 07:46 263阅读 0赞

## 1.新建javaSE工程--引入jar包 ##

![70][]

![70 1][]

## 2.开始写代码 ##

### 2.1 创建消息生产者 ###

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq;
 
 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;
 
 /** 
 * Title: MessageProducer 
 * Description:消息生产者 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageProducer { 
 	// 默认的连接用户名
 	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
 	// 默认的连接密码
 	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
 	// 默认的连接地址
 	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
 	// 发送的消息数量
 	private static final int SENDNUM=10; 
 	
 	public static void main(String[] args) {
 		//连接工厂
 		ConnectionFactory connectionFactory; 
 		//连接
 		Connection connection = null;
 		
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageProducer messageProducer ;//消息生产者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			
 			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
 			//创建消息队列
 			destination = session.createQueue("消息队列1");
 			//创建消息生产者
 			messageProducer = session.createProducer(destination);
 			//发送消息
 			sendMessage(session, messageProducer);
 			session.commit();
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		//关闭资源
 		finally {
 			if(null != connection)
 				try {
 					connection.close();
 				} catch (JMSException e) {
 					e.printStackTrace();
 				}
 		}
 		
 		
 	}
 	/**
 	 * 
 	 * Title: sendMessage 
 	 * Description: 发送消息 
 	 * @param session
 	 * @param messageProducer
 	 */
 	public static void sendMessage(Session session, MessageProducer messageProducer) {
 		try {
 			for (int i = 0; i < SENDNUM; i++) {
 				TextMessage textMessage = session.createTextMessage("ActiveMQ发送的第"+i+"条消息！");
 				System.out.println("发送消息"+i+"条！");
 				messageProducer.send(textMessage);
 			}
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 	}
 }

### 2.2 创建消息消费者--第一种方式 ###

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.ActiveMQConnectionFactory;
 
 /** 
 * Title: MessageProducer 
 * Description:消息消费者 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageConsumer { 
 	// 默认的连接用户名
 	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
 	// 默认的连接密码
 	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
 	// 默认的连接地址
 	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
 	// 发送的消息数量
 	private static final int SENDNUM=10; 
 	
 	public static void main(String[] args) {
 		//连接工厂
 		ConnectionFactory connectionFactory; 
 		//连接
 		Connection connection = null;
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageConsumer messageConsumer;//消息消费者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			//创建session
 			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
 			//创建消息队列
 			destination = session.createQueue("消息队列1");
 			//创建消息消费者
 			messageConsumer = session.createConsumer(destination);
 			//接收消息
 			while (true) {
 				TextMessage textMessage = (TextMessage)messageConsumer.receive(100000);
 				
 				if(null != textMessage) {
 					System.out.println("收到的消息是："+textMessage);
 				}else {
 					continue;
 				}
 				
 			}
 			
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		//关闭资源
 		finally {
 			if(null != connection)
 				try {
 					connection.close();
 				} catch (JMSException e) {
 					e.printStackTrace();
 				}
 		}
 		
 		
 	}
 }

### 2.3 **创建消息消费者--第二种方式（优化）--监听** ###

#### **2.3.1创建监听** ####

/** 
 * Title: MyListener.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
 import javax.jms.TextMessage;
 
 /** 
 * Title: MyListener 
 * Description:消息监听 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyListener implements MessageListener {
 
 	/* (non-Javadoc) 
 	 * Title: onMessage 
 	 * Description: 
 	 * @param message 
 	 * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
 	 */
 	@Override
 	public void onMessage(Message message) {
 		
 		try {
 			System.out.println("通过监听，监听收到的消息："+((TextMessage)message).getText());
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 
 	}
 
 }

#### 2.3.2 创建消费者--通过监听的方式进行消费 ####

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.ActiveMQConnectionFactory;
 
 /** 
 * Title: MessageProducer 
 * Description:消息消费者 --优化 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageConsumer2 { 
 	// 默认的连接用户名
 	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
 	// 默认的连接密码
 	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
 	// 默认的连接地址
 	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
 	// 发送的消息数量
 	private static final int SENDNUM=10; 
 	
 	public static void main(String[] args) {
 		//连接工厂
 		ConnectionFactory connectionFactory; 
 		//连接
 		Connection connection = null;
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageConsumer messageConsumer;//消息消费者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(MyMessageConsumer2.USERNAME, MyMessageConsumer2.PASSWORD, MyMessageConsumer2.BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			//创建session
 			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
 			//创建消息队列
 			destination = session.createQueue("消息队列1");
 			//创建消息消费者
 			messageConsumer = session.createConsumer(destination);
 			//接收消息--注册消息监听
 			messageConsumer.setMessageListener(new MyListener());
 			/*TextMessage textMessage = (TextMessage)messageConsumer.receive(100000);
 			System.out.println("第二个消费者，收到的消息是："+textMessage);*/
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		
 	}
 }

### 2.4 查询浏览器界面 ###

![70 2][]

![70 3][]

![70 4][]

[70]: /images/20220520/671331d1aadd47f0afeeff096f94f9c7.png
[70 1]: /images/20220520/afc8be3669684f48b93ad3edb2332624.png
[70 2]: /images/20220520/b289580488574aa0b30937ab49a37ce9.png
[70 3]: /images/20220520/9fc90c34749c4ab989fc13e7bd14c5ed.png
[70 4]: /images/20220520/cb3ff89bd83c4839a4000880c6ca3a1e.png