ActiveMQ （一）

---

 今天给你们分享的是ActiveMQ，若有不足，敬请指教。

 那么咱们必须知道ActiveMQ是什么。

1、ActiveMQ简介

1.1 ActiveMQ是什么

• ActiveMQ是一个消息队列应用服务器。支持JMS规范。

1.1.1 JMS概述

• 全称：Java Message Service ，即为Java消息服务，是一套java消息服务的API标准。（标准即接口）
• 实现了JMS标准的系统，称之为JMS Provider。

1.1.2 消息队列

1.1.2.1 概念

• 消息队列是在消息的传输过程当中保存消息的容器，提供一种不一样进程或者同一进程不一样线程直接通信的方式

图示

1. Producer：消息生产者，负责产生和发送消息到 Broker；
2. Broker：消息处理中心。负责消息存储、确认、重试等，通常其中会包含多个 queue；
3. Consumer：消息消费者，负责从 Broker 中获取消息，并进行相应处理；

1.2 ActiveMQ能作什么

• 实现两个不一样应用(程序)之间的消息通信。
• 实现同一个应用，不一样模块之间的消息通信。

1.3 ActiveMQ下载

• ActiveMQ官网地址： http://activemq.apache.org
• ActiveMQ下载地址：http://activemq.apache.org/download-archives.html

1.4 ActiveMQ主要特色

1. 支持多语言、多协议客户端。语言: Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议： OpenWire,Stomp REST,WS Notification,XMPP,AMQP
2. 对Spring的支持，ActiveMQ能够很容易整合到Spring的系统里面去。
3. 支持高可用、高性能的集群模式

---

2、示例

2.1 需求

• 使用ActiveMQ实现消息队列模型

2.2 配置步骤说明

1. 搭建ActiveMQ消息服务器（略）。
2. 建立一个java项目。
3. 建立消息生产者，发送消息。
4. 建立消息消费者，接收消息

2.3 第一部分 建立java项目，导入jar包

图示

2.4 第二部分 建立消息生成者，发送消息

2.4.1 建立MyProducer类，定义sengMsg2MQ方法

package com.xkt.producer;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * @author lzx
 *
 */
public class MyProducer {

	// 定义连接工厂
	private ConnectionFactory factory;
	// 定义连接
	private Connection connection;
	// 定义会话
	private Session session;
	// 定义目的地
	private Destination destination;
	// 定义消息
	private Message message;
	// 定义消息生生产者
	private MessageProducer producer;

	public void sengMsg2MQ(String msg) {

		try {
			/*
			 * 一、建立连接工厂
			 * 
			 * ActiveMQConnectionFactory(userName, password, brokerURL)
			 * 
			 * userName:用户名 默认admin password:密码 默认admin brokerURL：消息服务中心地址
			 * tcp://0.0.0.0:61616 基于tcp协议
			 */
			factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.109.3:61616");

			// 2.建立连接
			connection = factory.createConnection();
			// 开启连接
			connection.start();

			/*
			 * 三、建立会话
			 * 
			 * createSession(transacted, acknowledgeMode)
			 * 
			 * transacted:是否使用事物 true|false true 表示使用事物，每次对消息进行读写以后，要提交事物。若是使用了事物，则消息确认机制失效
			 * false 表示不使用事物
			 * 
			 * acknowledgeMode： 消息确认机制 Session.AUTO_ACKNOWLEDGE -
			 * 自动确认消息机制，一旦读取到消息，则消费成功，消息出队列，避免重复消费 Session.CLIENT_ACKNOWLEDGE -
			 * 客户端确认消息机制，手动确认，即消费了消息成功以后，再确认 Session.DUPS_OK_ACKNOWLEDGE -
			 * 有副本的客户端确认消息机制。集群模式下
			 * 
			 */
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

			// 4.建立队列
			destination = session.createQueue("test-mq");

			// 5.建立消息对象
			message = session.createTextMessage(msg);

			// 6.建立消息生产者
			producer = session.createProducer(destination);

			// 7.发送消息
			producer.send(message);

			// session.commit();

			System.out.println("消息发送成功");
		} catch (JMSException e) {
			e.printStackTrace();
			System.out.println("消息发送失败");
		} finally {
			// 回收消息发送者资源
			if (null != producer) {
				try {
					producer.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}

			// 回收会话资源
			if (null != session) {
				try {
					session.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}

			// 回收连接资源
			if (null != connection) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

}

2.4.2 建立一个测试类

package com.xkt.test;

import org.junit.Test;

import com.xkt.consumer.Myconsumer;
import com.xkt.producer.MyProducer;

/**
 * @author lzx
 *
 */
public class MessageTest {

	@Test
	public void testSend() {

		try {
			MyProducer producer = new MyProducer();
			producer.sengMsg2MQ("测试发送数据");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

2.4.3 测试

图示

• 查看ActiveMQ管理控制界面

图示

2.5 第三部分 建立消息消费者，消费消息

2.5.1 建立MyConsumer类

package com.xkt.consumer;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * @author lzx
 *
 */
public class Myconsumer {

	private ConnectionFactory factory;

	private Connection connection;

	private Session session;

	private Destination destination;

	private MessageConsumer consumer;

	private Message message;

	public void receiveFromMq() {

		try {
			factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://192.168.109.3:61616");

			connection = factory.createConnection();
			connection.start();

			session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

			// 建立目的地, 目的地命名即队列命名, 消息消费者须要经过此命名访问对应的队列
			destination = session.createQueue("queue");

			// 建立消息消费者, 建立的消息消费者与某目的地对应, 即方法参数目的地
			consumer = session.createConsumer(destination);

			// 六、读取消息
			message = consumer.receive(5000);

			// 7.提取文本
			if (null != message) {
				if (message instanceof TextMessage) {
					TextMessage tMsg = (TextMessage) message;
					String content = tMsg.getText();
					System.out.println("从列表中读取的是" + content);
				}
			}
			// 在手动确认机制下，消费完消息以后，必须手动确认，让消费的消息出队列不然，会出现重复消费的问题。
			message.acknowledge();

		} catch (JMSException e) {
			e.printStackTrace();
			System.out.println("读取失败");
		} finally {
			if (null != consumer) {
				try {
					consumer.close();
				} catch (JMSException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

			if (null != session) {
				try {
					session.close();
				} catch (JMSException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

			if (null != connection) {
				try {
					connection.close();
				} catch (JMSException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}

	}

}

2.5.2 修改测试类MessageTest，新增测试方法

@Test
	public void testReceive() {

		try {
			Myconsumer consumer = new Myconsumer();
			consumer.receiveFromMq();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

2.5.3 测试

图示

• 查看ActiveMQ管理控制界面

图示

---

  在前面的示例中，咱们发现消费者每次只能消费一条消息。当队列中有多条消息的时候，咱们须要屡次运行消费者，才能消费完这些消息。很麻烦！！！！如何解决这个问题呢？咱们将在后面的文章中给出。

版权说明：欢迎以任何方式进行转载，但请在转载后注明出处！