首页 编程语言正文

ActiveMQ简单的HelloWorld实例

符号 阅读：545 2021-03-31 22:11:45 评论：0

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

1、ActiviteMQ消息有3中形式

JMS 公共

点对点域

发布/订阅域

ConnectionFactory

QueueConnectionFactory

TopicConnectionFactory

Connection

QueueConnection

TopicConnection

Destination

Queue

Topic

Session

QueueSession

TopicSession

MessageProducer

QueueSender

TopicPublisher

MessageConsumer

QueueReceiver

TopicSubscriber

(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

发送消息的基本步骤：

(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包，项目截图:
  •  
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();   
	                }   
	    }   
}







标签:java编程语言
声明

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，请转载时务必注明文章作者和来源，不尊重原创的行为我们将追究责任；3.作者投稿可能会经我们编辑修改或补充。

相关文章
关注我们

一个IT知识分享的公众号

友情链接IM LINKS