ActiveMQ（三）——ActiviteMQ实现方式之二：Queue

发送方代码：

package com.mycom.activemq;

import java.util.HashMap;
import java.util.Map;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;

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

/**
 * ActiviteMQ方式2：Queue方式
 * 消息发送方（生产者）
 * 
 * @author guweiqiang
 */
public class QueueSender {

	/**
	 * 发送消息
	 */
	public static void sendMessage(String brokerUrl, Map<String, Object> map) {
		// QueueConnectionFactory : 连接工厂
		QueueConnectionFactory queueConnectionFactory;

		// QueueConnection : JMS客户端到JMS Provider的连接
		QueueConnection queueConnection = null;

		// QueueSession : 发送/接收消息的会话
		QueueSession queueSession = null;

		// Queue : 消息队列
		Queue queue;

		// QueueSender : 消息发送者
		javax.jms.QueueSender queueSender;

		try {
			// 创建一个连接工厂QueueConnectionFactory实例
			queueConnectionFactory = new ActiveMQConnectionFactory(
					ActiveMQConnection.DEFAULT_USER,
					ActiveMQConnection.DEFAULT_PASSWORD, brokerUrl);

			// 创建一个QueueConnection
			queueConnection = queueConnectionFactory.createQueueConnection();

			// 启动连接
			queueConnection.start();

			// 创建发送/接收消息的会话 QueueSession
			queueSession = queueConnection.createQueueSession(Boolean.TRUE,
					QueueSession.AUTO_ACKNOWLEDGE);

			// 创建消息队列 Queue
			queue = queueSession.createQueue("SecondQueue");

			// 创建消息发送者 QueueSender
			queueSender = queueSession.createSender(queue);
			queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

			// 准备工作已完成，可以开始发送消息了
			// 发送消息
			MapMessage mapMsg = queueSession.createMapMessage();
			mapMsg.setInt("ID", (Integer) map.get("id"));
			mapMsg.setString("NAME", (String) map.get("name"));
			mapMsg.setInt("AGE", (Integer) map.get("age"));
			System.out.println(mapMsg);
			queueSender.send(mapMsg);

			// 提交会话
			queueSession.commit();

		} catch (Exception ex) {
			ex.printStackTrace();
		} finally {
			// 释放资源
			if (queueSession != null) {
				try {
					queueSession.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if (queueConnection != null) {
				try {
					queueConnection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 测试
	 */
	public static void main(String[] args) {
		String brokerUrl = "tcp://127.0.0.1:61616";
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("id", 1002);
		map.put("name", "guweiqiang2");
		map.put("age", 22);
		QueueSender.sendMessage(brokerUrl, map);
	}

}

 

 

接收方代码：

package com.mycom.activemq;

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 org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * ActiviteMQ方式2：Queue方式
 * 消息接收方（消费者）
 * 
 * @author guweiqiang
 */
public class QueueReceiver {

	/**
	 * 接收消息
	 */
	public static void receiveMessage(String brokerUrl) {
		// QueueConnectionFactory : 连接工厂
		QueueConnectionFactory queueConnectionFactory;

		// QueueConnection : JMS客户端到JMS Provider的连接
		QueueConnection queueConnection = null;

		// QueueSession : 发送/接收消息的会话
		QueueSession queueSession = null;

		// Queue : 消息队列
		Queue queue;

		// QueueReceiver : 消息接收者
		javax.jms.QueueReceiver queueReceiver;

		try {
			// 创建一个连接工厂QueueConnectionFactory实例
			queueConnectionFactory = new ActiveMQConnectionFactory(
					ActiveMQConnection.DEFAULT_USER,
					ActiveMQConnection.DEFAULT_PASSWORD, brokerUrl);

			// 创建一个QueueConnection
			queueConnection = queueConnectionFactory.createQueueConnection();

			// 启动连接
			queueConnection.start();

			// 创建发送/接收消息的会话 QueueSession
			queueSession = queueConnection.createQueueSession(Boolean.TRUE,
					QueueSession.AUTO_ACKNOWLEDGE);

			// 创建消息队列 Queue
			queue = queueSession.createQueue("SecondQueue");

			// 创建消息接收者
			queueReceiver = queueSession.createReceiver(queue);

			// 准备工作已经完成，可以开始接收消息了（通过实现MessageListener的onMessage方法来接收消息）
			queueReceiver.setMessageListener(new MessageListener() {

				@Override
				public void onMessage(Message message) {
					if (message != null) {
						MapMessage mapMsg = (MapMessage) message;
						try {
							System.out.println("ID:" + mapMsg.getInt("ID")
									+ "\t NAME:" + mapMsg.getString("NAME")
									+ "\t AGE:" + mapMsg.getInt("AGE"));
						} catch (JMSException e) {
							e.printStackTrace();
						}
					}
				}
			});

			Thread.sleep(1000 * 100); // 进程睡眠一段时间再关闭

			// 提交会话
			queueSession.commit();

		} catch (Exception ex) {
			ex.printStackTrace();
		} finally {
			// 释放资源
			if (queueSession != null) {
				try {
					queueSession.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
			if (queueConnection != null) {
				try {
					queueConnection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 测试
	 */
	public static void main(String[] args) {
		String brokerUrl = "tcp://127.0.0.1:61616";
		QueueReceiver.receiveMessage(brokerUrl);
	}

}

 

 

启动ActiveMQ，再在本地执行上述发送方和接收方代码，运行结果如下：

发送方console：

ActiveMQMapMessage {commandId = 0, responseRequired = false, messageId = null, originalDestination = null, originalTransactionId = null, producerId = null, destination = null, transactionId = null, expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = false, type = null, priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false} ActiveMQMapMessage{ theTable = {NAME=guweiqiang2, AGE=22, ID=1002} }

 

 

接收方console：

ID:1002	 NAME:guweiqiang2	 AGE:22