ActiveMQ 实践之路(二) 使用Queue或者Topic发送/接受消息

cctvx1

浏览: 79892 次

最近访客更多访客>>

sparrow1314

hongyan1202

帅气的小码农

xuehanxin

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

ActiveMQ专栏

ActiveMQ JMS Apache Python Spring

本篇主要讲解在未使用其他框架(Spring)整合情况下,独立基于ActiveMQ,使用JMS规范进行消息通信。

     一.JMS回顾
       因为ActiveMQ是一个JMS Provider的实现,因此在开始实作前,有必要复习下JMS的基础知识
    Java Message Service (JMS)是sun提出来的为J2EE提供企业消息处理的一套规范,JMS目前有2套规范还在使用JMS 1.0.2b和1.1. 1.1已经成为主流的JMS Provider事实上的标准了.
      *1.1主要在session上面有一些重要改变,比如支持建立同一session上的transaction,让他支持同时发送P2P(Queue)消息和接受
Topic消息。

       在JMS中间主要定义了2种消息模式Point-to-Point (点对点),Publich/Subscribe Model (发布/订阅者)，
    其中在Publich/Subscribe 模式下又有Nondurable subscription和durable subscription (持久化订阅)2种消息处理方式。

     下面是JMS规范基本的接口和实现
     JMS Common Interfacse PTP-Specific Interface   Pub/Sub-specific interfaces
     ConnectionFactory     QueueConnectionFactory   TopicConnectionFactory
     Connection            QueueConnection          TopicConnection
     Destination           Queue                    Topic
     Session               QueueSession             TopiSession
     MessageProducer       QueueSender              TopicPublisher
     MessageConsumer       QueueReceiver/QueueBrwer TopicSubscriber

二.使用Queue

         下面以ActiveMQ example的代码为主进行说明

        1. 使用ActiveMQ的Connection，ConnectionFactory 建立连接,注意这里没有用到pool

java 代码

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

//建立Connection

java 代码

protected Connection createConnection() throws JMSException, Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, pwd, url);
Connection connection = connectionFactory.createConnection();
if (durable && clientID!=null) {
connection.setClientID(clientID);
}
connection.start();
return connection;
}

//建立Session

java 代码

protected Session createSession(Connection connection) throws Exception {
Session session = connection.createSession(transacted, ackMode);
return session;
}

2。发送消息的代码
//建立QueueSession

java 代码

protected MessageProducer createProducer(Session session) throws JMSException {
Destincation destination = session.createQueue("queue.hello");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
if( timeToLive!=0 )
producer.setTimeToLive(timeToLive);
return producer;
}

//使用Producer发送消息到Queue

java 代码

producer.send(message);

        3。接受消息,在JMS规范里面,你可以使用

java 代码

QueueReceiver/QueueBrowser直接接受消息,但是更多的情况下我们采用消息通知方式,即实现MessageListener接口
public void onMessage(Message message) {
//process message
}
//set MessageListner ,receive message
Destincation destination = session.createQueue("queue.hello");
consumer = session.createConsumer(destination);
consumer.setMessageListener(this);

以上就是使用jms queue发送接受消息的基本方式

三 Topic

1. 建立连接

java 代码

protected Connection createConnection() throws JMSException, Exception {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, pwd, url);
Connection connection = connectionFactory.createConnection();
//如果你要使用DurableSubScription 方式,你必须为connection设置一个ClientID
if (durable && clientID!=null) {
connection.setClientID(clientID);
}
connection.start();
return connection;
}

2. 建立Session

java 代码

protected Session createSession(Connection connection) throws Exception {
Session session = connection.createSession(transacted, ackMode);
return session;
}

3.创建Producer 发送消息到Topic

java 代码

//create topic on session
topic = session.createTopic("topic.hello");
producer = session.createProducer(topic);
//send message
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
producer.send(message);

4.创建Consumer接受消息(基本上和Queue相同)

java 代码

Destincation destination = session.createTopic("topic.hello");
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(this);
//如果你使用的是Durable Subscription方式,你必须在建立connection的时候
//设置ClientID,而且建立comsumer的时候使用createDurableSubscriber方法,为他指定一个consumerName。
//connection.setClientID(clientId);
//consumer = session.createDurableSubscriber((Topic) destination, consumerName);

四:连接ActiveMQ的方式
ActiveMQConnectionFactory 提供了多种连接到Broker的方式activemq.apache.org/uri-protocols.html

常见的有
vm://host:port     //vm
tcp://host:port    //tcp
ssl://host:port    //SSL
stomp://host:port //stomp协议可以跨语言,目前有很多种stomp client 库(java,c#,c/c++,ruby,python...);

分享到：

ActiveMQ 实践之路(三) ActiveMQ 4.x +JB ... | SpringSide2 MDP补充说明

2007-02-25 21:11
浏览 21525
评论(2)
论坛回复 / 浏览 (2 / 10489)
查看更多

2 楼 cctvx1 2007-02-27

其实ActiveMQ现在已经和作为很多项目的异步消息通信核心了
包括Mule,ServiceMix,Geronimo,同时它又进一步为Spring2.0提供了支持,这些在后续的文章里面都会陆续提到。

1 楼 SteveGY 2007-02-27

单独的MQ，用起来不太爽，如果和J2EE服务器一起使用，最好要一个RAR，但这类RAR还是比较麻烦的，虽然ActiveMQ提供了一个，但适用的AppServer都有限。
单独项目中使用带事务完整性支持的分布式队列，在工程上是有一定复杂度的，关键是事务完整性。

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论