面向消息的中间件:ActivaMQ及JMS

lnsoftware

2012-02-01

关注关注

Message-oriented Middleware (MOM).

ActiveMQ是一款使用JMS规范、用于系统之间远程通信的消息代理中间件。

ActiveMQ是一个MOM产品，针对应用程序提供了异步的、松散耦合的通信机制；而MOM产品遵循的是JMS规范，ActiveMQ 给系统架构提供了高可用性、高性能、高可扩展性、高稳定性和安全。

1. ActiveMQ特性

(1) JMS Compilance: 遵循JMS1.1规范；

(2) Connectivity:支持非常广泛的协议：HTTPS/S,IP多线传送，SSL,STOMP,TCP,UDP,XXMPP等；

(3) Pluggabel persistence and security: 提供可插拔的基于JDBC的的数据库访问及多种消息存储方式，以及可

插拔的权限配置；

(4) Buiding messaging application with java；

(5) Integration with application servers:可与Tomcat，Jetty, JBoss,Geronimo整合；

(6) ClientAPI:提供了基于Java,C/C++,.NET,Perl,PHP,Ruby的客户端API；

(7) Broker clustering

(8) Many advanced broker features and client options:例如支持Apache Camle;

(9) Dramatically(引人瞩目的) simplified administrator:Jconsole以及ActiveMQ WebConsole.

2. 适合应用ActiveMQ的场景

(1) 不同类型的应用系统集成

(2) 作为RPC的一个替换

(3) 减弱不同应用程序之间的耦合

(4) 作为Event-driven Archicture架构的支柱

(5) 改善应用程序的扩展性

3. Message-oriented Middleware

MOM是这样的一类软件：是一种异步的，松散耦合的，可靠的，可升级的，安全的，可在分布式系统之间通

讯的软件。

MOM可与JDBC进行类比，通过MOM访问其他系统就如通过JDBC访问DB一样。

像ACtiveMQ这样的JMS Broker的主要角色就是给Cleint Application提供一个通信的基础架构。

4. JMS Spec

JMS目的是提供一个标准的API，使用Java语言来发送和接受消息并且不受软件供应商的限制。

JMS 组件：

JMS规范定义了两中类型的Client：

JMS Client: JMS Client利用JMS API与JMS Provider进行交互，就如同使用JDBC API 与数据库进行交互一

样；JMS Clien使用MessageProducert发送消息和MessageConsumer接受消息。

Non-JMS Client: 一般是JMS Provider针对自己的产品提供了Client.

(1) JMS provider：JMS API的实现MOM；

(2) JMS message： JMS规范中最重要的概念，JMS规范中其他的概念都是围绕该概念而建立的；主要包括：

Header和Payload.

(3) JMS message header:JMSDestination,JMSDeliveryMode(Persistent,Nonpersistent),JMSExpiration,

JMSMessageID,JMSPriority(0-9),JMSTimestamp;

Client可配置的Header: JMSCorrelationID,JMSReplyTo,JMSType;

Provider可配置的Header: JMSRedelivered.

(4) JMS Message Properties:

public interface Message {
	...
	boolean getBooleanProperty(String name) throws JMSException;
	byte getByteProperty(String name) throws JMSException;
	short getShortProperty(String name) throws JMSException;
	int getIntProperty(String name) throws JMSException;
	long getLongProperty(String name) throws JMSException;
	float getFloatProperty(String name) throws JMSException;
	double getDoubleProperty(String name) throws JMSException;
	String getStringProperty(String name) throws JMSException;
	Object getObjectProperty(String name) throws JMSException;
	...
	Enumeration getPropertyNames() throws JMSException;
	boolean propertyExists(String name) throws JMSException;
	...
	void setBooleanProperty(String name, boolean value) throws JMSException;
	void setByteProperty(String name, byte value) throws JMSException;
	void setShortProperty(String name, short value) throws JMSException;
	void setIntProperty(String name, int value) throws JMSException;
	void setLongProperty(String name, long value) throws JMSException;
	void setFloatProperty(String name, float value) throws JMSException;
	void setDoubleProperty(String name, double value) throws JMSException;
	void setStringProperty(String name, String value) throws JMSException;
	void setObjectProperty(String name, Object value) throws JMSException;
	... 
}

(5) Meaaage Body:

JMS定义了6中类型的Message Body:

Message,TextMessage,MapMessage,BytesMessage,StreamMessage,ObjectMessage

5. Message selectors

用于过滤只想接受的消息，在Message Header中指定过滤条件，例如：

A JMS message with custom properties：

public void sendStockMessage(Session session,MessageProducer producer,
		Destination destination,String payload, String symbol,double price) throws JMSException {
	
	TextMessage textMessage = session.createTextMessage();
	textMessage.setText(payload);
	textMessage.setStringProperty("SYMBOL", symbol);
	textMessage.setDoubleProperty("PRICE", price);
	producer.send(destination, textMessage);
}

Filter messages using the SYMBOL header：

...
String selector = "SYMBOL = 'AAPL'";
MessageConsumer consumer =
session.createConsumer(destination, selector);
...

Filter messages using both the SYMBOL and PRICE headers：

...
String selector = "SYMBOL = 'AAPL' AND PRICE > " + getPreviousPrice();
MessageConsumer consumer =
session.createConsumer(destination, selector);
...

6. JMS Domain

(1) Point-ToPoint Domain

PTP Doamin使用的Destination是Queue, 消息可以同步也可以被异步的发送和接受，Queue中的消息有且只能

被Deliver一次。Consumer从Queue中接受消息同步情况下使用MessageConsumer.receive()方法，异步情况

通过MessageConsumer.setMessageListener()注册一个MessageListener实现。Queeu存储所有的消息直到她

们被Deliver或者过期。多个Consumer也可以注册道一个Queue上，但是每个Consumer只能接受指定的某类消

息。

(2) Publish/Subscribe Domain

Publish/Subscribe Domain使用的Destiantion是Topics, Publishers发送消息至Topic，Subscribers register接收

消息；任何发送至Topic的消息自动的Deliver给所有的Subscribers。同步情况下使用MessageConsumer.receive()

方法接收消息，异步情况下通过MessageConsumer.setMessageListener()注册一个MessageListener实现。默

认情况下，Topic部保存Message除非明确的指定，也可是使用durable subscription。使用durable subscription