2. JMS Provider(ActiveMQ)
特性及优势
1. 实现JMS1.1规范,支持J2EE1.4以上。
2. 可运行与任何JVM和大部分web容器(ActiveMQ works great in any JVM)
3. 支持多种语言客户端(java, C, C++, Ajax, ActionScript等等)
4. 支持多种协议(stomp, openwire, REST)
5. 良好的Spring支持(ActiveMQ has great Spring Support)
6. 速度很快,JBossMQ的十倍(ActiveMQ is very fast; often 10x faster than JBossMQ)
7. 与OpenJMS、JBossMQ等开源jms provider相比,ActiveMQ有apache的支持,持续发展的优势明显
Queue与Topic的比较
1. JMS Queue执行load balancer语义
一条消息仅能被一个consumer收到。如果在message发送的时候没有可用的consumer,那么它讲被保存一直到能处理该message的consumer可用。如果一个consumer收到一条message后却不响应它,那么这条消息将被转到另外一个consumer那儿。一个Queue可以有很多consumer,并且在多个可用的consumer中负载均衡。
2. Topic实现publish和subscribe语义
一条消息被publish时,他将发送给所有感兴趣的订阅者,所以零到多个subscriber将接收到消息的一个拷贝。但是在消息代理接收到消息时,只有激活订阅的subscriber能够获得消息的一个拷贝。
3. 分别对应两种消息模式
Point-to-Point(点对点),Publisher/Subscriber Model(发布/订阅者)
其中在Publicher/Subscriber模式下又有Nondurable subscription(非持久化订阅)和durable subscription(持久化订阅)两种消息处理方式。
Point-to-Point(点对点)消息模式开发流程
1. 生产者(producer)开发流程
1.1 创建 Connection
Java代码
// 根据url,user和password创建一个jms Connection。
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory (user, password, url);
connection = connectionFactory.createConnection();
connection.start();
1.2 创建Session
Java代码
/**在connection的基础上创建一个session,同时设置是否支持事务ACKNOWLEDGE标识。
• AUTO_ACKNOWLEDGE:自动确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收。
• CLIENT_ACKNOWLEDGE:客户端确认模式。会话对象依赖于应用程序对被接收的消息调用一个acknowledge()方法。一旦这个方法被调用,会话会确认最后一次确认之后所有接收到的消息。这种模式允许应用程序以一个调用来接收,处理并确认一批消息。注意:在管理控制台中,如果连接工厂的Acknowledge Policy(确认方针)属性被设置为"Previous"(提前),但是你希望为一个给定的会话确认所有接收到的消息,那么就用最后一条消息来调用acknowledge()方法。
• DUPS_OK_ACKNOWLEDGE:允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收;而且允许重复确认。在需要考虑资源使用时,这种模式非常有效。注意:如果你的应用程序无法处理重复的消息的话,你应该避免使用这种模式。如果发送消息的初始化尝试失败,那么重复的消息可以被重新发送。
• SESSION_TRANSACTED**/
Session session = connection.createSession(
transacted, Session.AUTO_ACKNOWLEDGE);
1.3 创建Destination对象
Java代码
//需指定其对应的主题(subject)名称,producer和consumer将根据subject来发送/接收对应的消息
if (topic) {
destination = session.createTopic(subject);
} else {
destination = session.createQueue(subject);
}
1.4 创建MessageProducer
Java代码
//根据Destination创建MessageProducer对象,同时设置其持久模式。
MessageProducer producer = session.createProducer(destination);
if (persistent) {
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
} else {
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}
if (timeToLive != 0) {
producer.setTimeToLive(timeToLive);
}
1.5 发送消息到队列(Queue)
Java代码
//封装TextMessage消息,使用MessageProducer的send方法将消息发送出去。
TextMessage message = session.createTextMessage("createMessageText");
producer.send(message);
2. 消费者(consumer)开发流程
2.1 实现MessageListener接口
Java代码
//消费者类必须实现MessageListener接口,然后在onMessage方法中监听消息到达处理。
2.2 创建Connection
Java代码
//根据url,user和password创建一个jms Connection,如果是durable模式,还需要给connection设置一个clientId。
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
Connection connection = connectionFactory.createConnection();
if (durable && clientId != null && clientId.length() > 0 && !"null".equals(clientId)) {
connection.setClientID(clientId);
}
connection.setExceptionListener(this);
connection.start();
2.3 创建Session和Destination
Java代码
//与产品类似
2.4 创建replayProducer【可选】
Java代码
//可以用来将消息处理结果发送给producer。
replyProducer = session.createProducer(null);
replyProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)
2.5 创建MessageConsumer
Java代码
//根据Destination创建MessageConsumer对象。
MessageConsumer consumer = null;
if (durable && topic) {
consumer = session.createDurableSubscriber((Topic)destination, consumerName);
} else {
consumer = session.createConsumer(destination);
}
2.6 消费message
Java代码
//在onMessage()方法中接收producer发送过来的消息进行处理,并可以通过replyProducer反馈信息给producer
if (message.getJMSReplyTo() != null) {
replyProducer.send(message.getJMSReplyTo(),
session.createTextMessage("Reply: " + message.getJMSMessageID()));
}
Publish/Subscriber(发布/订阅者)消息开发模式
1. 订阅者(Subscriber)开发流程
1.1 实现MessageListener接口
在onMessage()方法中监听发布者发出的消息队列,并做相应处理。
1.2 创建Connection
根据url, user和password创建一个jms connection
1.3 创建Session
在connection的基础上创建一个session,同时设置是否支持和ACKNOWLEDGE标志。
1.4 创建 Topic
创建两个Topic,topictest.message用于接收发布者发出的消息,topictest.control用于向发布者发送消息,实现双方的交互。
1.5 创建consumer和producer对象
根据topictest.message创建consumer,根据topictest.control创建producer
1.6 接收处理消息
在onMessage()方法中,对收到的消息进行处理,可直接简单在本地显示消息,或者根据消息内容不同处理对应的业务逻辑(比如:数据库更新、文件操作等等),并且可以使用 producer对象处理结果返回给发布者。
2. 发布者(Publisher)开发流程
2.1 实现MessageListener接口
在onMessage()方法中接收订阅者的反馈消息。
2.2 创建Connection
根据url, user和password 创建一个 jms Connection。
2.3 创建session
在connection的基础上创建一个session,同时设置是否支持事务和ACKNOWLEDGE标志。
2.4 创建Topic
创建两个Topic,topictest.messages用于向订阅者发布消息,topictest.control用于接收订阅者反馈的消息。这两个Topic与订阅者开发流程中的topic是一一对应的。
2.5 创建consumer和producer对象
根据topictest.message创建publisher;
根据topictest.control穿件consumer,同时监听订阅者反馈的消息。
3. JMS For Spring
Spring提供了用于简化JMS API使用的抽象框架,并且对用户屏蔽了JMS API中1.0.2和1.1版本的差异。
JMS的功能大致上分为两块,叫做消息制造和消息消耗。JmsTemplate用于制造消息和同步消息接收。和J2EE的事件驱动Bean风格类似,对于异步接收消息,Spring提供了一些消息监听容器来创建消息驱动的POJO(MDP)。
1. Spring 配置 connectionFactory
Xml代码
<!--客户端使用普通传输方式:tcp://localhost:61616
此处需加以注意的是Listener端的borkerURL使用了failover传输方式:
failover:(tcp://localhost:61616)?initialReconnectDelay=100&
maxReconnectAttempts=5
failover transport是一种重新连接机制,用于建立可靠的传输。此处配置的是一旦ActiveMQ broker中断,Listener端将每隔100ms自动尝试连接,直至成功连接或重试5次连接失败为止。
failover还支持多个borker同时提供服务,实现负载均衡的同时可增加系统容错性,格式: failover:(uri1,...,uriN)?transportOptions
-->
<bean id="jmsFactory" destroy-method="stop" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
<property name="userName" value="${activemq.username}" />
<property name="password" value="${activemq.password}" />
</bean>
</property>
</bean>
2. Spring JmsTemplate
Xml代码
<!-- Spring JMS Template
JmsTemplate 类的实例 一经配置便是线程安全 的。 要清楚一点,JmsTemplate
是有状态的,因为它维护了 ConnectionFactory 的引用,但这个状态时不是会话状态。
-->
<bean id="myJmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="jmsFactory" />
<property name="defaultDestinationName" value="MySubject" />
<!--JMS API有两种发送方法,一种采用发送模式、
优先级和存活时间作为服务质量(QOS)参数,
默认{deliveryMode:2(1),priority:4,timeToLive:0}
另一种使用无需QOS参数的缺省值方法。
<property name="explicitQosEnabled" value="true"/>
<property name="deliveryMode" value="2"/>
<property name="priority" value="4"/>
<property name="timeToLive" value="1000"/>
-->
<!--
<property name="receiveTimeout" value="1000" />
-->
</bean>
3. 发送的接收消息
Xml代码
<bean id="destinationTopic"
class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="HelloWorldTopic" />
</bean>
<!-- 读取信息 -->
<bean id="consumer" class="com.d1xn.jms.demo.Consumer">
<property name="jmsTemplate" ref="myJmsTemplate" />
</bean>
<!-- 发送信息 -->
<bean id="producerTopic" class="com.d1xn.jms.demo.ProducerTopic">
<property name="jmsTemplet" ref="myJmsTemplate" />
<property name="destination" ref="destinationTopic" />
</bean>
<!-- 消息监听 -->
<bean id="listenerContainerTopic"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsFactory" />
<property name="destination" ref="destinationTopic" />
<property name="messageListener" ref="consumer" />
<!—持久化客户端ID -->
<property name="clientId" value="clientId_001" />
<property name="subscriptionDurable" value="true" />
<property name="durableSubscriptionName" value="My_001" />
</bean>
说明(基于ActiveMQ5.4.2版本):
1、Web Console 的安全配置可参考
将conf/jetty.xml下面一段xml配置:
Xml代码
<bean id="securityConstraint" class="org.eclipse.jetty.http.security.Constraint">
<property name="name" value="BASIC" />
<property name="roles" value="admin" />
<property name="authenticate" value="true" />
</bean>
authenticate值设置为true即可!那用户名/密码的配置是在conf/jetty-realm.properties!
详细可参考
http://activemq.apache.org/web-console.html
2、连接安全配置可参考
将conf/activemq-security.xml中如下的配置
Xml代码
<plugins>
<!-- Configure authentication; Username, passwords and groups -->
<simpleAuthenticationPlugin anonymousAccessAllowed="false">
<users>
<authenticationUser username="system" password="${activemq.password}"
groups="admins"/>
<!--<authenticationUser username="user" password="${guest.password}"
groups="users"/>
<authenticationUser username="guest" password="${guest.password}" groups="guests"/>-->
</users>
</simpleAuthenticationPlugin>
</plugins>
copy至conf/activemq.xml中
Xml代码
<persistenceAdapter>
<kahaDB directory="${activemq.base}/data/kahadb"/>
</persistenceAdapter>
的下面(这是简单的用户名、密码的认证方式)!
用户名、密码的可在conf/credentials.properties配置!
详细可参考:
http://activemq.apache.org/security.html
http://activemq.apache.org/xml-reference.html
本文转载于:
http://boy00fly.iteye.com/blog/1103586
分享到:
相关推荐
ActiveMQ是中国最流行的开源消息中间件之一,由Apache软件基金会开发。它基于Java Message Service (JMS) 规范,提供了可靠的消息传递功能,适用于分布式系统中的应用间通信。本压缩包“activeMQ收发工具.rar”包含...
### ActiveMQ高并发处理方案详解 #### 一、引言 在现代分布式系统中,消息队列作为异步通信的核心组件之一,对于提高系统的吞吐量、降低响应时间和实现服务解耦等方面起着至关重要的作用。Apache ActiveMQ作为一款...
Apache ActiveMQ是开源的、基于Java消息服务(JMS)的应用服务器,它是Apache软件基金会的一部分。这个名为"apache-activemq-5.17.3"的压缩包包含了ActiveMQ的5.17.3版本,这是一个稳定且功能丰富的发布版本。在深入...
Apache ActiveMQ是业界广泛使用的开源消息中间件,它基于Java消息服务(JMS)标准,提供了高度可扩展、可靠的异步通信能力。标题"apache-activemq-5.16.5"指的是该软件的一个特定版本,即5.16.5版本,通常每个新版本...
本报告详细阐述了使用JMeter对ActiveMQ进行性能测试的过程和结果,旨在评估ActiveMQ在JMS(Java消息服务)环境下的性能表现。JMeter作为一个强大的负载和性能测试工具,被广泛用于测试各种应用程序,包括消息中间件...
jmx监控ActiveMQ监控 jmx(Java Management Extensions)是一种Java技术,为Java应用程序提供了管理和监控的功能。ActiveMQ是Apache软件基金会下的一个开源消息队列系统,提供了高效、可靠的消息传递服务。在生产...
在Linux环境下,Apache ActiveMQ是一个广泛使用的开源消息代理和队列服务器,它是Java Message Service (JMS) 的实现,能够处理大量的并发消息传递。ActiveMQ提供了高可用性、可扩展性和稳定性,使得它成为分布式...
Apache ActiveMQ是Apache软件基金会开发的一个开源消息中间件,它基于Java Message Service (JMS) 规范,提供高效、可靠的消息传递服务。在本文中,我们将深入探讨Apache ActiveMQ,特别是针对“apache-activemq-...
ActiveMQ路由配置方式 ActiveMQ路由配置是Apache ActiveMQ项目中的一种重要配置方式,它依赖另一个Apache项目Camel。ActiveMQ集成了Camel,启动时同时会启动Camel。通过Camel Web Console可以进行Routing配置。 ...
Apache ActiveMQ是开源社区中最流行的消息中间件之一,它基于Java消息服务(JMS)标准,提供高效、可靠的异步通信解决方案。ActiveMQ在企业级应用中广泛应用,因为它支持多种协议,如OpenWire、STOMP、AMQP、MQTT、...
ActiveMQ是中国最流行的开源消息中间件之一,基于Java Message Service(JMS)规范,它提供了一个高可伸缩、高性能、稳定且灵活的消息传递平台。这个"ActiveMQ接受和发送工具.rar"压缩包包含了用于与ActiveMQ交互的...
ActiveMQ是Apache软件基金会开发的一款开源消息中间件,它遵循开放消息传递标准(JMS,Java Message Service),用于在分布式系统中实现可靠的消息传递。在本文中,我们将深入探讨ActiveMQ v6.0.1的核心特性、应用...
ActiveMQ是一款非常流行的开源消息队列中间件,它实现了JMS(Java Message Service,Java消息服务)1.1规范,面向消息的中间件(Message Oriented Middleware,MOM)是指利用高效可靠的消息传递机制进行与平台无关的...
Spring集成ActiveMQ是将Spring框架与ActiveMQ消息中间件相结合,实现异步处理和解耦应用程序的关键技术。在本文中,我们将深入探讨如何配置和使用这一组合,以及它在实际项目中的应用。 首先,让我们了解Spring框架...
#### 一、ActiveMQ简介 ActiveMQ是一款非常流行的开源消息中间件,它基于Java语言开发,并且遵循了Java消息服务(JMS)规范。ActiveMQ提供了丰富的特性,包括持久化消息存储、事务支持、集群等功能。在企业级应用中,...
ActiveMQ in Action pdf英文原版加源代码压缩包。 Apache ActiveMQ in Action is a thorough, practical guide to implementing message-oriented systems in Java using ActiveMQ. The book lays out the core of ...
Apache ActiveMQ是业界广泛使用的开源消息中间件,尤其在Linux环境下表现出色。它基于Java语言开发,遵循Apache软件基金会的许可证,并且实现了多种消息传递协议,包括OpenWire、STOMP、AMQP和XMPP等。在Linux系统上...
Apache ActiveMQ是开源社区中最流行的Java消息代理,也是企业级消息中间件(Message Broker)的首选之一。在最新的稳定版5.15.0中,它提供了可靠的消息传递功能,适用于分布式应用程序之间的通信,实现了异步处理、...
本资源提供的内容是关于ActiveMQ的连接池实现,分为两部分:一是作者自己实现的ActiveMQ连接池,二是新版本ActiveMQ自带的连接池。连接池是一种资源管理技术,通过复用已建立的数据库连接或网络连接,减少创建和销毁...
而ActiveMQ是Apache出品的一款开源消息中间件,它遵循JMS(Java Message Service)规范,用于处理应用程序之间的异步通信。本教程将详细介绍如何在Spring Boot项目中集成ActiveMQ,实现消息接收的Demo。 首先,我们...