ActiveMQ实战

  ActiveMQ是JMS规范的一个实现产品,即对JMS的一套接口进行了实现。所以下面我们先对JMS与ActiveMQ进行一个简单介绍,然后再介绍ActiveMQ的P2P与PubSub两种开发模式。

一、    JMS简介

JMS即Java消息服务,是J2EE的13规范之一,是一套面向消息中间件(MOM)的应用程序接口。用在两个程序之间,或分布式系统中发送消息,进行异步通信。

JMS使我们能够通过消息收发服务(也可以称为,消息中介程序或路由器)从一个JMS客户机向另一个JMS客户机发送消息。消息是JMS中的一种类型对象,由两部分组成:报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载的类型来划分,消息可分为多种,它们分别携带:简单文本消息、可序列化的对象、属性集合、字节流、原始值流、无有效负载消息。

二、    ActiveMQ简介

ActiveMQ出身名门,是Apache门下的最流行的,能力强劲的开源消息总线。完全支持JMS1.1和J2EE1.4规范的JMS Provide实现。它从设计上保证了高性能的集群,当然实现了JMS的P2P与PubSub两种开发模式。

三、    安装运行ActiveMQ

首先下载ActiveMQ的稳定版:http://activemq.apache.org ,如下图,我下载的是apache-activemq-5.11.1-bin:

  

然后进入到bin/win64文件夹下,并启动activemq.bat:

  

出现如下提示表示ActiveMQ启动成功:

  

然后在浏览器输入:http://localhost:8161/即可看到ActiveMQ的管理界面:

  

然后我们点击:Manage ActiveMQ broker登录管理后台:(用户名、密码默认都为admin)

  

登录进去以后我们可以点击Queues看到消息队列以及消费者情况等:

  

ActiveMQ的环境已经好了,下面我们开始开发P2P与PubSub的实例。

四、    P2P开发模式实战

建立JavaProject:ActiveMQ-p2p,并导入ActiveMQ的核心jar包,只需要导入根目录下的activemq-all.jar即可:

  

然后建立一个消息生产者类:JMSProducer;一个消息消费者类:JMSConsumer:

源码如下:

消息生产者类:JMSProducer

package com.tgb.jms.ActiveMQ;

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

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

/**
 * 消息生产者
 *
 * @author Administrator
 *
 */
public class JMSProducer {

	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
	private static final int SENDNUM = 10; // 发送的消息数量

	public static void main(String[] args) {

		ConnectionFactory connectionFactory; // 连接工厂
		Connection connection = null; // 连接
		Session session; // 会话 接受或者发送消息的线程
		Destination destination; // 消息的目的地
		MessageProducer messageProducer; // 消息生产者

		// 实例化连接工厂
		connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);

		try {
			connection = connectionFactory.createConnection(); // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session,第一个参数为是否开启事务
			destination = session.createQueue("FirstQueue1"); // 创建消息队列
			messageProducer = session.createProducer(destination); // 创建消息生产者
			sendMessage(session, messageProducer); // 发送消息
			session.commit();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 发送消息
	 *
	 * @param session
	 * @param messageProducer
	 * @throws Exception
	 */
	public static void sendMessage(Session session, MessageProducer messageProducer) throws Exception {
		for (int i = 0; i < JMSProducer.SENDNUM; i++) {
			TextMessage message = session.createTextMessage("ActiveMQ 发送的消息" + i);
			System.out.println("发送消息:" + "ActiveMQ 发送的消息" + i);
			messageProducer.send(message);
		}
	}
}

消息消费者类:JMSConsumer

package com.tgb.jms.ActiveMQ;

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

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

/**
 * 消息消费者
 *
 * @author wangzhipeng
 *
 */
public class JMSConsumer {
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址

	public static void main(String[] args) {
		ConnectionFactory connectionFactory; // 连接工厂
		Connection connection = null; // 连接
		Session session; // 会话 接受或者发送消息的线程
		Destination destination; // 消息的目的地
		MessageConsumer messageConsumer;// 消息消费者

		// 实例化连接工厂
		connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);

		try {
			connection = connectionFactory.createConnection();// 通过工厂获取连接
			connection.start();// 启动连接
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 第一个参数为是否开启事务
			destination = session.createQueue("FirstQueue1");// 创建消息队列
			messageConsumer = session.createConsumer(destination);// 创建消息消费者
			/*
			 * 实际应用中,不会这么用,会注册一个监听
			 */
			while (true) {
				TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
				if (textMessage != null) {
					System.out.println("收到的消息:" + textMessage.getText());
				} else {
					break;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

然后我们运行JMSProducer的main方法来生产10条消息,成功执行后,我们刷新一下后台页面,可以看到我们刚才创建的消息队列:FirstQueue1,消息个数为10,没有消费者,进入队列的消息为10,出列消息个数为0:

  

然后我们运行JMSConsumer类的main方法,来接收FirstQueue1的消息,成功执行后,我们再刷新后台页面,发现消息已被接收:

  

  P2P开发模式优化

  上面的实例中我们接收消息的方式为,写了一个死循环while (true) 来不停的接收消息,这样很浪费cpu资源,实际生产中不会这么做。这里,我们可以注册一个监听器,当监听到有消息入队列后,才去接收消息。所以,改动如下:

  首先,新建一个监听类Listener,需要实现MessageListener接口:

package com.tgb.jms.ActiveMQ2;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * 消息监听
 *
 * @author zhipeng
 *
 */
public class Listener implements MessageListener {

	@Override
	public void onMessage(Message message) {
		// TODO Auto-generated method stub
		try {
			System.out.println("收到的消息:" + ((TextMessage) message).getText());
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}

  然后我们将消息接收者类JMSConsumer的while (true)死循环代码替换为messageConsumer.setMessageListener

(newListener());即可,如下图:

  

  这样一来我们就符合我们人为的思维操作了,会节省很多cpu资源。

五、    PubSub开发模式实战

P2P的模式相当于一个消息生产者一个消费者,PubSub中可以有一个消息发布者多个消息订阅者,开发与P2P几乎一样,只是将createQueue(创建消息队列)改为了createTopic(创建主题);生产消息改为了发布消息;接收消息改为了订阅消息。代码这里就不再贴出来了。有需要的同志到这里下载http://download.csdn.net/detail/wang379275614/9023177

这里需要注意的是,订阅者必须先订阅,然后发布者发送消息后,订阅者才能自动收到消息。

六、    总结

ActiveMQ是JMS规范的一个实现产品,主要用在两个程序之间,或分布式系统中发送消息,进行异步通信。我们可以用它来解决高并发的问题,或者分布式事务的问题等。

版权声明:本文为博主原创文章,未经博主允许不得转载。

时间: 2024-11-09 02:48:37

ActiveMQ实战的相关文章

ActiveMQ实战篇之ActiveMQ实现request/reply模型(二)

ActiveMQ实战篇之ActiveMQ实现request/reply模型(二) 原文地址:https://www.cnblogs.com/bincoding/p/8254113.html

ActiveMQ的作用总结(应用场景及优势)以及springboot+activeMq 实战

业务场景说明: 消息队列在大型电子商务类网站,如京东.淘宝.去哪儿等网站有着深入的应用, 队列的主要作用是消除高并发访问高峰,加快网站的响应速度. 在不使用消息队列的情况下,用户的请求数据直接写入数据库,在高并发的情况下,会对数据库造成巨大的压力,同时也使得系统响应延迟加剧. 在使用队列后,用户的请求发给队列后立即返回, (例如: 当然不能直接给用户提示订单提交成功,京东上提示:您"您提交了订单,请等待系统确认"), 再由消息队列的消费者进程从消息队列中获取数据,异步写入数据库. 由于

Spring下ActiveMQ实战

MessageQueue是分布式的系统里经常要用到的组件,一般来说,当需要把消息跨网段.跨集群的分发出去,就可以用这个.一些典型的示例就是: 1.集群A中的消息需要发送给多个机器共享: 2.集群A中消息需要主动推送,但彼此的网络不是互通的(如集群A只有过HA才能被外界访问): 当然上面的几个点,除了用MQ还有其它实现方式,但是MQ无疑是非常适合用来做这些事的.众多MQ中,ActiveMQ是比较有名气也很稳定的,它发送消息的成本非常廉价,支持Queue与Topic两种消息机制.本文主要就是讲如何在

Java消息队列--ActiveMq 实战

原文地址:http://www.cnblogs.com/jaycekon/p/6225058.html 1.下载安装ActiveMQ ActiveMQ官网下载地址:http://activemq.apache.org/download.html ActiveMQ 提供了Windows 和Linux.Unix 等几个版本,楼主这里选择了Linux 版本下进行开发. 下载完安装包,解压之后的目录: 从它的目录来说,还是很简单的: bin存放的是脚本文件 conf存放的是基本配置文件 data存放的是

ActiveMQ实战-集群

原文:http://blog.csdn.net/lifetragedy/article/details/51869032 ActiveMQ的集群 内嵌代理所引发的问题: 消息过载 管理混乱 如何解决这些问题--集群的两种方式: Master slave Broker clusters ActiveMQ的集群有两种方式: MASTER/SLAVE模式 Cluster模式 Pure Master Slave Pure master slave的工作方式: 当master broker失效的时候.Sl

Apache ActiveMQ实战(2)-集群

ActiveMQ的集群 内嵌代理所引发的问题: 消息过载 管理混乱 如何解决这些问题--集群的两种方式: Master slave Broker clusters ActiveMQ的集群有两种方式: MASTER/SLAVE模式 Cluster模式 Pure Master Slave Pure master slave的工作方式: 当master broker失效的时候.Slave broker 做出了两种不同的相应方式 启动network connectors和transport connec

Apache ActiveMQ实战(1)-基本安装配置与消息类型

ActiveMQ简介 ActiveMQ是一种开源的,实现了JMS1.1规范的,面向消息(MOM)的中间件,为应用程序提供高效的.可扩展的.稳定的和安全的企业级消息通信.ActiveMQ使用Apache提供的授权,任何人都可以对其实现代码进行修改. ActiveMQ的设计目标是提供标准的,面向消息的,能够跨越多语言和多系统的应用集成消息通信中间件.ActiveMQ实现了JMS标准并提供了很多附加的特性.这些附加的特性包括,JMX管理(java Management Extensions,即java

activemq 实战二 连接到ActiveMQ-Connecting to ActiveMQ

The main role of a JMS broker such as ActiveMQ is to provide a communication infrastructure for client applications. For that reason, ActiveMQ provides connectors, a connectivity mechanism that provides client-to-broker communications (using transpor

activemq 实战 四 传输连接器-Transport connectors 4.2

In order to exchange messages, producers and consumers (clients) need to connect to the broker. This client-to-broker communication is performed through transport connectors. ActiveMQ provides an impressive list of protocols clients can use to exchan