笔者带你剖析轻量级消息中间件ActiveMQ

《笔者带你剖析轻量级消息中间件ActiveMQ》

 

前言

软件系统的发展，简单来说是从单机应用逐步演变为分布式+集群模式，当然这根互联网的兴起是密不可分的。笔者曾经面试过一些开发人员，当你们使用代码编写一个简单的Login模块时，大概需要花费多少周期？有些开发人员不假思索的脱口而出，5分钟、10分钟等。OK，我们先不管他们是否真的能够在这么短的时间内开发出一个简单的Login功能，我真正想问的是，这个看似简单的功能背后，如果存放在互联网场景下，如何能够保证系统同时承载100——100W人的并发洗礼而不会出现宕机？紧接着是死一般的沉寂。

 

目录

一、分布式系统与消息中间件

二、JMS与ActiveMQ

三、ActiveMQ的下载与启动

四、消息的传递模型

五、Queue与Topic

六、补充——消息流程

 

一、分布式系统与消息中间件

前言这个故事，或许并没有笔者说得这么夸张，但仔细想想看，如果业务场景对并发性、实时性要求较高时，任何一个看似简单的系统功能，都将会变得异常复杂，因为这里的复杂是来源于多方面的，最主要的就是如果系统因为某一个功能出现异常将会导致整个系统服务不可用。那么应该怎么办呢？如果你不想过多的，或者是在短时间内追加过多的投资，集群+物理堆机+适当的代码性能调优，或许是解决问题的最好办法，但这样真的一劳永逸吗？我们都知道集群的威力从理论上来说会是单节点部署的数倍，但集群如果没有细化，也就是说，一个系统的所有业务模块仍然耦合在一起时，无论你的集群的规模有多么庞大，从外部看起来，仍然只是一个大杂烩，凌乱不堪。

 

图a 单节点部署与集群部署

 

那么如果是在条件允许的情况下，真正该做的事情是根据不同的业务逻辑，将原本冗余在一起的业务模块打散、拆分成多个子系统，然后进行模块化部署，每一个业务子系统都可以是一个集群，这样一来，我们的系统才会变得更加健壮和高效。当然，分布式环境有分布式环境的好处，但是，同样也有它的弊端，毕竟系统依赖多了，无论是部署还是调试或者是排查问题的时候都会变得无比的复杂和麻烦，本章笔者并不打算教会大家如何根据具体的业务场景构建一个健壮的大型分布式平台，因此请回到本章的主题——MQ身上。

 

MQ(Message Queue，消息队列)，一切以消息为主。假设你现在已经将原本冗余的一个系统根据业务逻辑进行了适当的拆分，并且每一个子系统都是一个集群，在这种情况下，由UM子系统处理跟用户相关的所有业务，但当它处理完自身业务后，还需要调用其它子系统的服务时应该怎么办？或许某些开发人员会觉得这似乎很简单，直接调用啊，一个长/短链接请求就可以解决。但是这样一来，UM子系统将会被动的依赖N多的子系统，如果后期再进行追加，UM的依赖会越来越多，并且这种同步调用的方式在某些情况下似乎并不高效。换句话多，UM子系统的职责仅仅只是负责与用户相关的所有业务，其它业务与它无关，哪怕是服务调用这个过程也不该由它来做，如果你也是这么想的，那么恭喜你，你已经考虑到跨系统之间的依赖解耦和通过异步调用的方式提升程序的响应性能的高度。

 图b 直接服务调用与MQ消息推送

 

如图b所示，UM子系统的任务非常明确，当处理完用户登录任务后，直接将登录结果发送给MQ服务器，如果订阅了该主题的子系统自然会收到消息，接下来，各个子系统该干嘛干嘛去，UM子系统彻底解耦。这里有一点笔者需要提出来，本章的主角ActiveMQ或许并不太适用于互联网场景，至于为什么，笔者稍后会进行详解。

 

二、JMS与ActiveMQ

长话短说，JMS(Java Message Service，Java消息服务)属于JAVA EE规范之一，同JDBC规范一样，JMS也仅仅只提供有通用性的接口，而具体的实现，则需要依赖由其遵循了JMS规范的消息中间件负责。ActiveMQ是Apache旗下的一款完全支持JMS1.1和JAVA EE 1.4规范的消息中间件产品，换句话说，ActiveMQ就是JMS规范的实现者之一。

 

之前笔者曾经提及过，ActiveMQ或许并不太适用于互联网场景，因为JMS规范注定了ActiveMQ更适合于企业级项目。在大型的分布式互联网场景下，跨系统之间的解耦和异步调用这2点，仅仅只是属于任何一款消息中间件产品最基础的特征，而大部分情况下，开发人员却往往更加关注消息的顺序、扩展性、可靠性、消息发送的一致性，以及多集群之间订阅等方面的问题。因此，在大型的分布式互联网场景下，开发人员更多会选择MetaQ、Notify、kafka等天生为互联网场景而生的消息中间件产品。或许你会觉得奇怪，为什么笔者会拿ActiveMQ进行开刀？其实最主要的原因是因为其简单、轻量、易部署的特性非常适合作为笔者消息中间件系列博文的开篇来进行阐述。

 

三、ActiveMQ的下载与启动

所谓工欲善其事必先利其器，ActiveMQ的下载地址为http://activemq.apache.org/download.html，本章笔者所使用的版本为apache-activemq-5.11.1(for Windows)，当然笔者建议大家下载跟本文一致的版本，这样既可避免一些不必要的错误发生。

 

当成功下载好ActiveMQ后，接下来要做的事情有2个，第一是在目录“\apache-activemq-5.11.1\bin”中启动我们的消息服务器(JMS Provider)， 启动成功就可以访问管理员界面：http://localhost:8161/admin，默认用户名和密码admin/admin。如果你想修改用户名和密码的话，在conf/jetty-realm.properties中修改即可。另外一个便是将目录“apache-activemq-5.11.1\lib”中的所有构件拷贝到项目工程中，以便于实现MQ客户端。

 

四、消息传递模型

之前笔者曾经提及过，ActiveMQ其实就是JMS规范的一种实现，那么按照JMS的架构模型来看，JMS Provider的主要任务就是负责消息路由，推送消息的客户端称之为生产者(Producer)，而接受消息的客户端则称之为消费者(Consumer)。JMS的消息传递模型有2种，一种是基于点对点(Point-to-Point，p2p)的，而另外一种则是基于广播形式的发布/订阅(publish/subscribe，pub/sub)。

 

点对点的消息传递比较简单，无非就是一种一对一的消息推送/消费模式，也就是说，如果多个客户端都在监听消息队列上的消息，JMS Provider则会根据先到先得原则确定唯一的一个消费者，如果没有客户端在监听消息队列，那么消息将会保存在消息队列中，直到最终确定消费者为止，点对点模型其实就是一个典型的拉模型(而非Provider主动推送)。发布/订阅可以理解为一对多的消息推送模式，在发布/订阅模型中，只要订阅了目标主题的所有客户端都可以消费生产者推送的消息，pub/sub就是典型的推模型，由JMS Provider主动推送消息。

图c JMS的消息传递模型

 

这里笔者还要补充一个内容，JMS的消息体主要由3部分构成，分别是：头(head)、属性(property)，以及主体(body)。主体就是具体的消息内容，当然对应不同的消息类型，JMS提供有不同的消息接口，这些消息接口都属于javax.jms.Message接口，比如：StreamMessage、MapMessage、TextMessage、ObjectMessage、BytesMessage等。

 

五、Queue与Topic

说了这么多，是该用代码进行验证的时候了，那么接下来笔者将演示如何使用ActiveMQ实现p2p和pub/sub等2种消息传递模型。在JMS规范中，Queue实现了基于P2P的消息传递模型，而Topic则实现了基于pub/sub的消息传递模型。

 

使用Queue实现基于P2P的消息传递模型：

/**
 * p2p模型生产者
 * 
 * @author gaoxianglong
 */
public class Producer {
	private ConnectionFactory connFactory;
	private Connection conn;
	private Session session;
	private MessageProducer producer;

	Producer() {
		try {
			/* 创建链接工厂 */
			connFactory = new ActiveMQConnectionFactory("admin", "admin",
					"tcp://127.0.0.1:61616");
			/* 获取链接 */
			conn = connFactory.createConnection();
			conn.start();
			/* 获取Session */
			session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
			/* 创建生产者 */
			producer = session.createProducer(session.createQueue("testQueue"));
			producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			System.out.println(sendMessage() ? "消息发送成功..." : "消息发送成功...");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (null != conn) {
				try {
					conn.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public boolean sendMessage() {
		boolean result = false;
		if (null != session && null != producer) {
			try {
				/* 发送消息 */
				producer.send(session.createTextMessage("test..."));
				session.commit();
				result = true;
			} catch (JMSException e) {
				try {
					session.rollback();
				} catch (JMSException e1) {
					e1.printStackTrace();
				}
			}
		}
		return result;
	}

	public static void main(String[] args) {
		new Producer();
	}
}

 

/**
 * Queue消费者
 *
 * @author gaoxianglong
 */
public class Consumer implements MessageListener {
	private ConnectionFactory connFactory;
	private Connection conn;
	private Session session;
	private MessageConsumer consumer;

	Consumer() {
		try {
			connFactory = new ActiveMQConnectionFactory("admin", "admin",
					"tcp://127.0.0.1:61616");
			conn = connFactory.createConnection();
			conn.start();
			session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
			consumer = session.createConsumer(session.createQueue("testQueue"));
			consumer.setMessageListener(this);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		new Consumer();
	}

	@Override
	public void onMessage(Message arg0) {
		TextMessage msg = (TextMessage) arg0;
		try {
			System.out.println("收到消息->" + msg.getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
}

 

上述程序示例了，笔者使用Queue实现了基于P2P的消息传递模型，而如果希望使用Topic实现pub/sub模型，只需要将生产者/消费者的代码片段：

session.createQueue("testQueue")

 

 更改为如下即可：

session.createTopic("testTopic")

 

六、补充——消息流程

 发送消息的基本步骤：
1、创建连接工厂类；
2、通过ConnectionFactory建立连接Connection，并启动；
3、使用连接Connection建立会话Session；
4、使用会话Session和管理对象Destination创建消息生产者MessageSender；
5、生产者发送消息；

 

消息接收者从JMS接受消息的步骤
1、创建连接工厂类；
2、通过ConnectionFactory建立连接Connection，并启动；
3、使用连接Connection建立会话Session；
4、使用会话Session和管理对象Destination创建消息接收者MessageReceiver；
5、消费者消费消息；

 

本章内容到此结束，由于时间仓库，本文或许有很多不尽人意的地方，希望各位能够理解和体谅。另外http://boy00fly.iteye.com/blog/1103586也是一篇不错的介绍ActiveMQ的博文。