ActiveMQ简单的HelloWorld实例

清疚 2023-10-16 22:49 55阅读 0赞

在上一篇博文[JMS 规范简介][JMS]，我们介绍了JMS的两种消息模型：点对点和发布订阅模型，以及消息被消费的两个方式：同步和异步，JMS编程模型的对象，最后说了JMS的优点。这篇博文我们主要介绍ActiveMQ是如何实现JMS规范中的两种消费模式。

> 1、ActiviteMQ消息有3中形式
> 
> <table style="border:1px solid silver; border-collapse:collapse; word-break:break-word"> 
> <tbody> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> JMS&nbsp;公共 </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> 点对点域 </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> 发布/订阅域 </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> ConnectionFactory </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> QueueConnectionFactory </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> TopicConnectionFactory </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> Connection </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> QueueConnection </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> TopicConnection </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> Destination </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> Queue </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> Topic </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> Session </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> QueueSession </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> TopicSession </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> MessageProducer </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> QueueSender </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> TopicPublisher </td> 
> </tr> 
> <tr> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> MessageConsumer </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> QueueReceiver </td> 
> <td style="color:rgb(69,69,69); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; border:1px solid silver; border-collapse:collapse; padding:3px"> TopicSubscriber </td> 
> </tr> 
> </tbody> 
> </table>
> 
> (1)、点对点方式（point-to-point）
> 
> 点对点的消息发送方式主要建立在 Message Queue,Sender,reciever上，Message Queue 存贮消息，Sneder 发送消息，receive接收消息.具体点就是Sender Client发送Message Queue ,而 receiver Cliernt从Queue中接收消息和"发送消息已接受"到Quere,确认消息接收。消息发送客户端与接收客户端没有时间上的依赖，发送客户端可以在任何时刻发送信息到Queue，而不需要知道接收客户端是不是在运行
> 
> (2)、发布/订阅 方式（publish/subscriber Messaging）
> 
> 发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式，可能存在多个接收客户端，并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法，destination的receive方法，和实现message listener 接口的onMessage 方法。

> 2、ActiviteMQ接收和发送消息基本流程
> 
> [![image][]][image 1]
> 
> **发送消息的基本步骤：**
> 
> (1)、创建连接使用的工厂类JMS ConnectionFactory
> 
> (2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动
> 
> (3)、使用连接Connection 建立会话Session
> 
> (4)、使用会话Session和管理对象Destination创建消息生产者MessageSender
> 
> (5)、使用消息生产者MessageSender发送消息
> 
> **消息接收者从JMS接受消息的步骤**
> 
> (1)、创建连接使用的工厂类JMS ConnectionFactory
> 
> (2)、使用管理对象JMS ConnectionFactory建立连接Connection，并启动
> 
> (3)、使用连接Connection 建立会话Session
> 
> (4)、使用会话Session和管理对象Destination创建消息接收者MessageReceiver
> 
> (5)、使用消息接收者MessageReceiver接受消息，需要用setMessageListener将MessageListener接口绑定到MessageReceiver消息接收者必须实现了MessageListener接口，需要定义onMessage事件方法。

#### 搭建开发环境 ####

*  建立项目  
    我们只需要建立一个java项目就可以了，导入jar包，项目截图:
 *  ![Center][]

1、JMS方式发送接收消息:

JMS发送者
 
 package com.mq.producer;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
 import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.ActiveMQConnectionFactory;
 
 
 
 public class MessageSender {
 	// 发送次数
 	public static final int SEND_NUM = 5;
 	// tcp 地址
 	public static final String BROKER_URL = "tcp://localhost:61616";
 	// 目标，在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
 	public static final String DESTINATION = "mq.queue";
 
 	/**
 	 * 发送消息
 	 * 
 	 * @param session
 	 * @param producer
 	 * @throws Exception
 	 */
 	public static void sendMessage(Session session, MessageProducer producer) throws Exception {
 		for (int i = 0; i < SEND_NUM; i++) {
 			String message = "发送消息第" + (i + 1) + "条";
 			TextMessage text = session.createTextMessage(message);
 
 			System.out.println(message);
 			producer.send(text);
 		}
 	}
 
 	public static void run() throws Exception {
 
 		Connection connection = null;
 		Session session = null;
 		try {
 			// 创建链接工厂
 			ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
 					ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);
 			// 通过工厂创建一个连接
 			connection = factory.createConnection();
 			// 启动连接
 			connection.start();
 			// 创建一个session会话
 			session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
 			// 创建一个消息队列
 			Destination destination = session.createQueue(DESTINATION);
 			// 创建消息制作者
 			MessageProducer producer = session.createProducer(destination);
 			// 设置持久化模式
 			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 			sendMessage(session, producer);
 			// 提交会话
 			session.commit();
 
 		} catch (Exception e) {
 			throw e;
 		} finally {
 			// 关闭释放资源
 			if (session != null) {
 				session.close();
 			}
 			if (connection != null) {
 				connection.close();
 			}
 		}
 	}
 	
 
 	
 	
 
 }

JMS 接收者
    package com.mq.consumer;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class MessageReceiver {
    	// tcp 地址
    	public static final String BROKER_URL = "tcp://localhost:61616";
    	// 目标，在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
    	public static final String DESTINATION = "mq.queue";
    
    	public static void run() throws Exception {
    
    		Connection connection = null;
    		Session session = null;
    		try {
    			// 创建链接工厂
    			ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    					ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);
    			// 通过工厂创建一个连接
    			connection = factory.createConnection();
    			// 启动连接
    			connection.start();
    			// 创建一个session会话
    			session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    			// 创建一个消息队列
    			Destination destination = session.createQueue(DESTINATION);
    			// 创建消息制作者
    			MessageConsumer consumer = session.createConsumer(destination);
    
    			while (true) {
    				// 接收数据的时间（等待） 100 ms
    				Message message = consumer.receive(1000 * 100);
    
    				TextMessage text = (TextMessage) message;
    				if (text != null) {
    					System.out.println("接收：" + text.getText());
    				} else {
    					break;
    				}
    			}
    
    			// 提交会话
    			session.commit();
    
    		} catch (Exception e) {
    			throw e;
    		} finally {
    			// 关闭释放资源
    			if (session != null) {
    				session.close();
    			}
    			if (connection != null) {
    				connection.close();
    			}
    		}
    	}
    }

JMS 测试工具类
    package com.mq.test;
    
    import com.mq.consumer.MessageReceiver;
    import com.mq.producer.MessageSender;
    
    public class CommonTest {
    
    	public static void main(String[] args) {
    		// TODO Auto-generated method stub
    		try {
    			MessageSender.run();
    			Thread.sleep(10*1000);
    			MessageReceiver.run();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    }

2、Queue队列方式发送点对点消息数据

发送者
 package com.mq.producer;
 
 import javax.jms.DeliveryMode;
 import javax.jms.MapMessage;
 import javax.jms.Queue;
 import javax.jms.QueueConnection;
 import javax.jms.QueueConnectionFactory;
 import javax.jms.QueueSession;
 import javax.jms.Session;
 
 import org.apache.activemq.ActiveMQConnection;
 import org.apache.activemq.ActiveMQConnectionFactory;
 
 public class QueueSender {
 	// 发送次数
 	public static final int SEND_NUM = 5;
 	// tcp 地址
 	public static final String BROKER_URL = "tcp://localhost:61616";
 	// 目标，在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
 	public static final String DESTINATION = "activemq";
 
 	/**
 	 * 发送消息
 	 * 
 	 * @param session
 	 * @param sender
 	 * @throws Exception
 	 */
 	public static void sendMessage(QueueSession session, javax.jms.QueueSender sender) throws Exception {
 		for (int i = 0; i < SEND_NUM; i++) {
 			String message = "发送消息第" + (i + 1) + "条";
 
 			MapMessage map = session.createMapMessage();
 			map.setString("text", message);
 			map.setLong("time", System.currentTimeMillis());
 			System.out.println(map);
 
 			sender.send(map);
 		}
 	}
 
 	/**
 	 * Queue 发送信息核心方法
 	 * 
 	 * @throws Exception
 	 */
 	public static void run() throws Exception {
 
 		QueueConnection connection = null;
 		QueueSession session = null;
 		try {
 			// 创建链接工厂
 			QueueConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
 					ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);
 			// 通过工厂创建一个连接
 			connection = factory.createQueueConnection();
 			// 启动连接
 			connection.start();
 			// 创建一个session会话
 			session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
 			// 创建一个消息队列
 			Queue queue = session.createQueue(DESTINATION);
 			// 创建消息发送者
 			javax.jms.QueueSender sender = session.createSender(queue);
 			// 设置持久化模式
 			sender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 			sendMessage(session, sender);
 			// 提交会话
 			session.commit();
 
 		} catch (Exception e) {
 			throw e;
 		} finally {
 			// 关闭释放资源
 			if (session != null) {
 				session.close();
 			}
 			if (connection != null) {
 				connection.close();
 			}
 		}
 	}
 }

接收者
    package com.mq.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.Queue;
    import javax.jms.QueueConnection;
    import javax.jms.QueueConnectionFactory;
    import javax.jms.QueueSession;
    import javax.jms.Session;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class QueueReceiver {
    	// tcp 地址
    	public static final String BROKER_URL = "tcp://localhost:61616";
    	// 目标，在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
    	public static final String TARGET = "activemq";
    
    	/**
    	 * @throws Exception
    	 */
    	public static void run() throws Exception {
    
    		QueueConnection connection = null;
    		QueueSession session = null;
    		try {
    			// 创建链接工厂
    			QueueConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
    					ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);
    			// 通过工厂创建一个连接
    			connection = factory.createQueueConnection();
    			// 启动连接
    			connection.start();
    			// 创建一个session会话
    			session = connection.createQueueSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    			// 创建一个消息队列
    			Queue queue = session.createQueue(TARGET);
    			// 创建消息制作者
    			javax.jms.QueueReceiver receiver = session.createReceiver(queue);
    
    			receiver.setMessageListener(new MessageListener() {
    				public void onMessage(Message msg) {
    					if (msg != null) {
    						MapMessage map = (MapMessage) msg;
    						try {
    							System.out.println(map.getLong("time") + "接收#" + map.getString("text"));
    						} catch (JMSException e) {
    							e.printStackTrace();
    						}
    					}
    				}
    			});
    			// 休眠100ms再关闭
    			Thread.sleep(1000 * 10);
    
    			// 提交会话
    			session.commit();
    
    		} catch (Exception e) {
    			throw e;
    		} finally {
    			// 关闭释放资源
    			if (session != null) {
    				session.close();
    			}
    			if (connection != null) {
    				connection.close();
    			}
    		}
    	}
    }

Queue测试工具类
    package com.mq.test;
    
    import com.mq.consumer.QueueReceiver;
    import com.mq.producer.QueueSender;
    
    public class QueueTest {
    	public static void main(String[] args){
    		try {
    			QueueSender.run();
    			Thread.sleep(10*1000);
    			QueueReceiver.run();
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    
    }

3、 Topic主题发布和订阅消息

发送者
 package com.mq.producer;
 
 import javax.jms.Connection; 
 import javax.jms.DeliveryMode; 
 import javax.jms.JMSException; 
 import javax.jms.MessageProducer; 
 import javax.jms.Session; 
 import javax.jms.TextMessage; 
 import javax.jms.Topic; 
 import org.apache.activemq.ActiveMQConnectionFactory; 
 
 public class TopicPublisher {
 	 public static void main(String[] args) throws JMSException { 
 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); 
 Connection connection = factory.createConnection(); 
 connection.start(); 
 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
 Topic topic = session.createTopic("myTopic.messages"); 
 MessageProducer producer = session.createProducer(topic); 
 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); 
 for(int i =0;i<10;i++){ 
 TextMessage message = session.createTextMessage(); 
 message.setText("message_" + System.currentTimeMillis()); 
 producer.send(message); 
 System.out.println("Sent message: " + message.getText()); 
 try { 
 Thread.sleep(1000); 
 } catch (InterruptedException e) { 
 e.printStackTrace(); 
 } 
 } 
 // session.close(); 
 // connection.stop(); 
 // connection.close(); 
 } 
 
 }

接收者
    package com.mq.consumer;
    
    import javax.jms.Connection;  
    import javax.jms.JMSException;  
    import javax.jms.Message;  
    import javax.jms.MessageConsumer;  
    import javax.jms.MessageListener;  
    import javax.jms.Session;  
    import javax.jms.TextMessage;  
    import javax.jms.Topic;    
    import org.apache.activemq.ActiveMQConnectionFactory;  
    
    public class TopicSubscriber {
    	 public static void main(String[] args) {  
    	        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
    	                try {  
    	                    Connection connection = factory.createConnection();  
    	                    connection.start();  
    	                    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
    	                    Topic topic = session.createTopic("myTopic.messages");  
    	                    MessageConsumer consumer = session.createConsumer(topic);  
    	                    consumer.setMessageListener(new MessageListener() {  
    	                        public void onMessage(Message message) {  
    	                            TextMessage tm = (TextMessage) message;  
    	                            try {  
    	                                System.out.println("Received message: " + tm.getText());  
    	                            } catch (JMSException e) {  
    	                                e.printStackTrace();  
    	                            }  
    	                        }  
    	                    });  
    	                } catch (JMSException e) {  
    	                    e.printStackTrace();  
    	                }  
    	    }  
    }

[JMS]: http://blog.csdn.net/zhouzhiwengang/article/details/50441247
[image]: http://images.cnitblog.com/blog/151517/201306/27085728-78e9c08eff7d45b3b6d6d92fd7d50f23.png
[image 1]: http://images.cnitblog.com/blog/151517/201306/27085725-1265cef3c9664ea2b5e1eaf3c1ade03a.png
[Center]: https://img-blog.csdn.net/20151231145757580?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQv/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center