消息隊列ActiveMQ的使用詳解

時間  2019-11-07
標籤 消息 隊列 activemq 使用 詳解 欄目 ActiveMQ 简体版
原文   原文鏈接

經過上一篇文章 《消息隊列深刻解析》，咱們已經消息隊列是什麼、使用消息隊列的好處以及常見消息隊列的簡單介紹。java

這一篇文章，主要帶你們詳細瞭解一下消息隊列ActiveMQ的使用。spring

學習消息隊列ActiveMQ的使用以前，咱們先來搞清JMS。apache

JMS服務器

  1. JMS基本概念 JMS（JAVA Message Service,java消息服務）是java的消息服務，JMS的客戶端之間能夠經過JMS服務進行異步的消息傳輸。JMS（JAVA Message Service,java消息服務）API是一個消息服務的標準或者說是規範，容許應用程序組件基於JavaEE平臺建立、發送、接收和讀取消息。它使分佈式通訊耦合度更低，消息服務更加可靠以及異步性。session

  2. JMS五種不一樣的消息正文格式 JMS定義了五種不一樣的消息正文格式，以及調用的消息類型，容許你發送並接收以一些不一樣形式的數據，提供現有消息格式的一些級別的兼容性。app

StreamMessage – Java原始值的數據流 MapMessage–一套名稱-值對 TextMessage–一個字符串對象 ObjectMessage–一個序列化的 Java對象 BytesMessage–一個字節的數據流 3.JMS兩種消息模型 1 .點到點（P2P）模型異步

使用隊列（Queue）做爲消息通訊載體；知足生產者與消費者模式，一條消息只能被一個消費者使用，未被消費的消息在隊列中保留直到被消費或超時。好比：咱們生產者發送100條消息的話，兩個消費者來消費通常狀況下兩個消費者會按照消息發送的順序各自消費一半（也就是你一個我一個的消費。）後面咱們會經過代碼演示來驗證。socket

  1. 發佈/訂閱（Pub/Sub）模型

發佈訂閱模型（Pub/Sub） 使用主題（Topic）做爲消息通訊載體，相似於廣播模式；發佈者發佈一條消息，該消息經過主題傳遞給全部的訂閱者，在一條消息廣播以後才訂閱的用戶則是收不到該條消息的。tcp

4.JMS編碼接口之間的關係分佈式

ConnectionFactory：建立Connection對象的工廠，針對兩種不一樣的jms消息模型，分別有QueueConnectionFactory和TopicConnectionFactory兩種。能夠經過JNDI來查找ConnectionFactory對象。

Connection：Connection表示在客戶端和JMS系統之間創建的連接（對TCP/IP socket的包裝）。Connection能夠產生一個或多個Session。跟ConnectionFactory同樣，Connection也有兩種類型：QueueConnection和TopicConnection。

Session：Session是操做消息的接口。能夠經過session建立生產者、消費者、消息等。Session提供了事務的功能。當須要使用session發送/接收多個消息時，能夠將這些發送/接收動做放到一個事務中。一樣，也分QueueSession和TopicSession。

MessageProducer：消息生產者由Session建立，並用於將消息發送到Destination。一樣，消息生產者分兩種類型：QueueSender和TopicPublisher。能夠調用消息生產者的方法（send或publish方法）發送消息。

MessageConsumer ：消息消費者由Session建立，用於接收被髮送到Destination的消息。兩種類型：QueueReceiver和TopicSubscriber。可分別經過session的createReceiver(Queue)或createSubscriber(Topic)來建立。固然，也能夠session的creatDurableSubscriber方法來建立持久化的訂閱者。

Destination：Destination的意思是消息生產者的消息發送目標或者說消息消費者的消息來源。對於消息生產者來講，它的Destination是某個隊列（Queue）或某個主題（Topic）;對於消息消費者來講，它的Destination也是某個隊列或主題（即消息來源）。

MessageListener： 消息監聽器。若是註冊了消息監聽器，一旦消息到達，將自動調用監聽器的onMessage方法。

參考：blog.csdn.net/shaobingj12…

消息隊列ActiveMQ 1.簡介 ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個徹底支持JMS1.1和J2EE 1.4規範的 JMS Provider實現,儘管JMS規範出臺已是好久的事情了,可是JMS在當今的J2EE應用中間仍然扮演着特殊的地位。

2.簡單使用 添加Maven依賴

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.15.3</version>
</dependency>
複製代碼

2.1.測試點對點模型通訊 生產者發送消息測試方法：

@Test
public void testQueueProducer() throws Exception {

	// 一、建立一個鏈接工廠對象，須要指定服務的ip及端口。
	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.155:61616");
	// 二、使用工廠對象建立一個Connection對象。
	Connection connection = connectionFactory.createConnection();
	// 三、開啓鏈接，調用Connection對象的start方法。
	connection.start();
	// 四、建立一個Session對象。
	// 第一個參數：是否開啓事務。若是true開啓事務，第二個參數無心義。通常不開啓事務false。
	// 第二個參數：應答模式。自動應答或者手動應答。通常自動應答。
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	// 五、使用Session對象建立一個Destination對象。兩種形式queue、topic，如今應該使用queue
	Queue queue = session.createQueue("test-queue");
	// 六、使用Session對象建立一個Producer對象。
	MessageProducer producer = session.createProducer(queue);
	// 七、建立一個Message對象，可使用TextMessage。
	for (int i = 0; i < 50; i++) {
		TextMessage textMessage = session.createTextMessage("第"+i+ "一個ActiveMQ隊列目的地的消息");
		// 八、發送消息
		producer.send(textMessage);
	}

	// 九、關閉資源
	producer.close();
	session.close();
	connection.close();
}
複製代碼

消費者消費消息測試方法

@Test
public void testQueueConsumer() throws Exception {
	// 建立一個ConnectionFactory對象鏈接MQ服務器
	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.155:61616");
	// 建立一個鏈接對象
	Connection connection = connectionFactory.createConnection();
	// 開啓鏈接
	connection.start();
	// 使用Connection對象建立一個Session對象
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	// 建立一個Destination對象。queue對象
	Queue queue = session.createQueue("test-queue");
	// 使用Session對象建立一個消費者對象。
	MessageConsumer consumer = session.createConsumer(queue);
	// 接收消息
	consumer.setMessageListener(new MessageListener() {

		@Override
		public void onMessage(Message message) {
			// 打印結果
			TextMessage textMessage = (TextMessage) message;
			String text;
			try {
				text = textMessage.getText();
				System.out.println("這是接收到的消息：" + text);
			} catch (JMSException e) {
				e.printStackTrace();
			}

		}
	});
	// 等待接收消息
	System.in.read();
	// 關閉資源
	consumer.close();
	session.close();
	connection.close();
}
複製代碼

咱們開啓兩個消費者進程來監聽(運行兩次testQueueConsumer()方法)。

而後咱們運行運行生產者測試方法發送消息.先發送消息仍是先監聽消息通常不會不影響。

效果以下:

兩個消費者各自消費一半消息，並且仍是按照消息發送到消息隊列的順序，這也驗證了咱們上面的說法。 第一個消費者

第二個消費者

2.2.測試發佈/訂閱（Pub/Sub）模型通訊 生產者發送消息測試方法：

@Test
public void testTopicProducer() throws Exception {
	// 一、建立一個鏈接工廠對象，須要指定服務的ip及端口。
	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.155:61616");
	// 二、使用工廠對象建立一個Connection對象。
	Connection connection = connectionFactory.createConnection();
	// 三、開啓鏈接，調用Connection對象的start方法。
	connection.start();
	// 四、建立一個Session對象。
	// 第一個參數：是否開啓事務。若是true開啓事務，第二個參數無心義。通常不開啓事務false。
	// 第二個參數：應答模式。自動應答或者手動應答。通常自動應答。
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	// 五、使用Session對象建立一個Destination對象。兩種形式queue、topic，如今應該使用topic
	Topic topic = session.createTopic("test-topic");
	// 六、使用Session對象建立一個Producer對象。
	MessageProducer producer = session.createProducer(topic);
	// 七、建立一個Message對象，可使用TextMessage。
	for (int i = 0; i < 50; i++) {
		TextMessage textMessage = session.createTextMessage("第"+i+ "一個ActiveMQ隊列目的地的消息");
		// 八、發送消息
		producer.send(textMessage);
	}
	// 九、關閉資源
	producer.close();
	session.close();
	connection.close();
}
複製代碼

消費者消費消息測試方法：

@Test
public void testTopicConsumer() throws Exception {
	// 建立一個ConnectionFactory對象鏈接MQ服務器
	ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.25.155:61616");
	// 建立一個鏈接對象
	Connection connection = connectionFactory.createConnection();
	// 開啓鏈接
	connection.start();
	// 使用Connection對象建立一個Session對象
	Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
	// 建立一個Destination對象。topic對象
	Topic topic = session.createTopic("test-topic");
	// 使用Session對象建立一個消費者對象。
	MessageConsumer consumer = session.createConsumer(topic);
	// 接收消息
	consumer.setMessageListener(new MessageListener() {

		@Override
		public void onMessage(Message message) {
			// 打印結果
			TextMessage textMessage = (TextMessage) message;
			String text;
			try {
				text = textMessage.getText();
				System.out.println("這是接收到的消息：" + text);
			} catch (JMSException e) {
				e.printStackTrace();
			}

		}
	});
	System.out.println("topic消費者啓動。。。。");
	// 等待接收消息
	System.in.read();
	// 關閉資源
	consumer.close();
	session.close();
	connection.close();
}
複製代碼

先運行兩個消費者進程（提早訂閱，否則收不到發送的消息），而後運行生產者測試方法發送消息。

結果是： 兩個消費者進程均可以接收到生產者發送過來的全部消息，我這裏就不貼圖片了， 這樣驗證了咱們上面的說法。

咱們從上面代碼就能夠看出，點對點通訊和發佈訂閱通訊模式的區別就是建立生產者和消費者對象時提供的Destination對象不一樣，若是是點對點通訊建立的Destination對象是Queue，發佈訂閱通訊模式通訊則是Topic。

3.整合Spring使用 整合spring除了咱們上面依賴的Jar包還要依賴

<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-jms</artifactId>
	<version>4.2.7.RELEASE</version>
</dependency>
<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-context-support</artifactId>
	<version>4.2.7.RELEASE</version>
 </dependency>
複製代碼

好比咱們在咱們的系統中如今有兩個服務，第一個服務發送消息，第二個服務接收消息，咱們下面看看這是如何實現的。

發送消息 發送消息的配置文件：

<!-- 真正能夠產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
	<property name="brokerURL" value="tcp://192.168.25.155:61616" />
</bean>
<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
	class="org.springframework.jms.connection.SingleConnectionFactory">
	<!-- 目標ConnectionFactory對應真實的能夠產生JMS Connection的ConnectionFactory -->
	<property name="targetConnectionFactory" ref="targetConnectionFactory" />
</bean>
<!-- 配置生產者 -->
<!-- Spring提供的JMS工具類，它能夠進行消息發送、接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
	<!-- 這個connectionFactory對應的是咱們定義的Spring提供的那個ConnectionFactory對象 -->
	<property name="connectionFactory" ref="connectionFactory" />
</bean>
<!--這個是隊列目的地，點對點的 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
	<constructor-arg>
		<value>spring-queue</value>
	</constructor-arg>
</bean>
<!--這個是主題目的地，一對多的 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
	<constructor-arg value="topic" />
</bean>
複製代碼

發送消息的測試方法：

@Test public void testSpringActiveMq() throws Exception { //初始化spring容器 ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-activemq.xml"); //從spring容器中得到JmsTemplate對象 JmsTemplate jmsTemplate = applicationContext.getBean(JmsTemplate.class); //從spring容器中取Destination對象 Destination destination = (Destination) applicationContext.getBean("queueDestination"); //使用JmsTemplate對象發送消息。 jmsTemplate.send(destination, new MessageCreator() {

@Override
		public Message createMessage(Session session) throws JMSException {
			//建立一個消息對象並返回
			TextMessage textMessage = session.createTextMessage("spring activemq queue message");
			return textMessage;
		}
	});
}
複製代碼

咱們上面直接ApplicationContext的getBean方法獲取的對象，實際在項目使用依賴注入便可。

接收消息 建立一個MessageListener的實現類。

public class MyMessageListener implements MessageListener {

@Override
public void onMessage(Message message) {
	
	try {
		TextMessage textMessage = (TextMessage) message;
		//取消息內容
		String text = textMessage.getText();
		System.out.println(text);
	} catch (JMSException e) {
		e.printStackTrace();
	}
}
複製代碼

}

接收消息的配置文件：

<!-- 真正能夠產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
	<property name="brokerURL" value="tcp://192.168.25.168:61616" />
</bean>
<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory"
	class="org.springframework.jms.connection.SingleConnectionFactory">
	<!-- 目標ConnectionFactory對應真實的能夠產生JMS Connection的ConnectionFactory -->
	<property name="targetConnectionFactory" ref="targetConnectionFactory" />
</bean>
<!--這個是隊列目的地，點對點的 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
	<constructor-arg>
		<value>spring-queue</value>
	</constructor-arg>
</bean>
<!--這個是主題目的地，一對多的 -->
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
	<constructor-arg value="topic" />
</bean>
<!-- 接收消息 -->
<!-- 配置監聽器 -->
<bean id="myMessageListener" class="cn.e3mall.search.listener.MyMessageListener" />
<!-- 消息監聽容器 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
	<property name="connectionFactory" ref="connectionFactory" />
	<property name="destination" ref="queueDestination" />
	<property name="messageListener" ref="myMessageListener" />
</bean>
複製代碼

測試接收消息的代碼

@Test
public void testQueueConsumer() throws Exception {
	
	//初始化spring容器
	ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-activemq.xml");
	//等待
	System.in.read();
}
複製代碼
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程