不为有趣之事,何遣有涯之生
不失其所者久,死而不亡者寿

AMQ简明教程(5) JMS的API开发步骤

JMS的API开发步骤

JMS开发的基本步骤

1:创建一个JMS connection factory
2:通过connection factory来创建JMS connection
3:启动JMS connection
4:通过connection创建JMS session
5:创建JMS destination
6:创建JMS producer,或者创建JMS message,并设置destination
7:创建JMS consumer,或者是注册一个JMS message listener
8:发送或者接受JMS message(s)
9:关闭所有的JMS资源(connection, session, producer, consumer等)

下面是JMS的API结构图

对于非持久的Topic消息的发送

基本跟前面发送队列信息是一样的,只是把创建Destination的地方,由创
建队列替换成创建Topic,例如:

Destination destination = session.createTopic("MyTopic");

对于非持久的Topic消息的接收

1:必须要接收方在线,然后客户端再发送信息,接收方才能接收到消息
2:同样把创建Destination的地方,由创建队列替换成创建Topic,例如:

Destination destination = session.createTopic("MyTopic");

3:由于不知道客户端发送多少信息,因此改成while循环的方式了,例如:

Message message = consumer.receive();
while(message!=null) {
TextMessage txtMsg = (TextMessage)message;
System.out.println("收到消息:" + txtMsg.getText());
message = consumer.receive(1000L);
} 

下面是一个详细例子:

package jmstest.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Created by hzqiuxm on 2015/5/30.
 * 非持久topic消息发送
 */
public class topicSend {

    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.55:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();

        //开启事务,消息确认方式为自动
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        //创建队列
        Destination destination = session.createTopic("my-topic");
        //生产者把消息传送到队列中
        MessageProducer producer = session.createProducer(destination);
        for(int i=0; i<3; i++) {
            TextMessage message = session.createTextMessage("message no persister--"+i);
//            Thread.sleep(1000);
            //通过消息生产者发出消息
            producer.send(message);
        }
        session.commit();
        session.close();
        connection.close();
    }
}
package jmstest.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Created by hzqiuxm on 2015/5/30.
 * 非持久的topic接收
 */
public class topicReceive {

    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.55:61616");
        Connection connection = cf.createConnection();
        connection.start();
        //开启事务,

        final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic("my-topic");
        MessageConsumer consumer = session.createConsumer(destination);

        Message message = consumer.receive();
        while(message!=null) {
            TextMessage txtMsg = (TextMessage)message;
            System.out.println("收到消息:" + txtMsg.getText());
            message = consumer.receive(1000L);
        }
        session.commit();
        session.close();
        connection.close();
    }
}

先启动生产者,再启动消费者,会发现消费者接受不到消息。
进入MQ的界面中查看:

消息有3条,有一个消费者,但消费的消息为0

再执行一次生产者,进行查看

对于持久的Topic消息的发送

ConnectionFactory connectionFactory = new
ActiveMQConnectionFactory("tcp://192.168.1.106:61616");
Connection connection = connectionFactory.createConnection();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Topic destination = session.createTopic("MyTopic");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
connection.start();
for(int i=0; i<2; i++) {
TextMessage message = session.createTextMessage("messagedd--"+i);
Thread.sleep(1000);
//通过消息生产者发出消息
producer.send(message);
}
session.commit();
session.close();
connection.close();

1:要用持久化订阅,发送消息者要用DeliveryMode.PERSISTENT 模式发现,在连接之前设定
2:一定要设置完成后,再start 这个connection

对于持久的Topic消息的接收

ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.106:61616");
Connection connection = cf.createConnection();
connection.setClientID("cc1");
final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Topic destination = session.createTopic("MyTopic");
TopicSubscriber ts = session.createDurableSubscriber(destination, "T1");
connection.start();
Message message = ts.receive();
while(message!=null) {
TextMessage txtMsg = (TextMessage)message;
session.commit();
System.out.println("收到消息:" + txtMsg.getText());
message = ts.receive(1000L);
}
session.close();
connection.close();

1:需要在连接上设置消费者id,用来识别消费者
2:需要创建TopicSubscriber来订阅
3:要设置好了过后再start 这个connection
4:一定要先运行一次,等于向消息服务中间件注册这个消费者,然后再运行客户端发送信息,这个时候,
无论消费者是否在线,都会接收到,不在线的话,下次连接的时候,会把没有收过的消息都接收下来。

详细例子:

package jmstest.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Created by Administrator on 2015/6/1.
 * 持久化topic发送者
 */
public class perTopicSend {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.1.55:61616");
        Connection connection = connectionFactory.createConnection();

        //开启事务,消息确认方式为自动
        Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        //创建队列
        Destination destination = session.createTopic("my-topic");
        //生产者把消息传送到队列中
        MessageProducer producer = session.createProducer(destination);

        //设置持久化投递方式后再启动
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        connection.start();

        for(int i=0; i<3; i++) {
            TextMessage message = session.createTextMessage("message persister11--"+i);
            //通过消息生产者发出消息
            producer.send(message);
        }
        session.commit();
        session.close();
        connection.close();
    }
}
package jmstest.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * Created by Administrator on 2015/6/1.
 * 持久化topic接收者,要优先生产者运行一次,相当于在MQ上注册一个订阅消息
 */
public class perTopicReceive {
    public static void main(String[] args) throws Exception {
        ConnectionFactory cf = new ActiveMQConnectionFactory("tcp://192.168.1.55:61616");
        Connection connection = cf.createConnection();
        //设置消费者的身份标示
        connection.setClientID("qxm001");

        //开启事务
        final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
        //设置订阅topic并创建订阅
        Topic destination = session.createTopic("my-topic");
        TopicSubscriber ts = session.createDurableSubscriber(destination,"T1");

        connection.start();

        Message message = ts.receive();
        while(message!=null) {
            TextMessage txtMsg = (TextMessage)message;
            session.commit();
            System.out.println("收到消息:" + txtMsg.getText());
            message = ts.receive(1000L);
        }

        session.close();
        connection.close();
    }
}

首先运行消费者订阅消息

然后停止消费者

我们可以发现在topic和subscribers上可以查看到该消费者,尽管消费者此时已经离线

然后运行消息生产者,查看topic列表,发现消息数目增加了

再次启动消费者:

发现subscribers里显示了接收的消息,但是topic接收消息的DE对列还是未变

这是因为订阅模式中,topic中的消息可以被很多消费者订阅并接收

不管是持久化还是非持久化订阅,都是要消费者先运行才可以!

持久化消息

这是ActiveMQ 的默认传送模式,此模式保证这些消息只被传送一次和成功使用一次。
对于这些消息,可靠性是优先考虑的因素。可靠性的另一个重要方面是确保持久性消息传送至目标后,消息服务在向消费者传送它们之前不会丢失这些消息。
这意味着在持久性消息传送至目标时,消息服务将其放入持久性数据存储。
如果消息服务由于某种原因导致失败,它可以恢复此消息并将此消息传送至相应的消费者。
虽然这样增加了消息传送的开销,但却增加了可靠性。

非持久化消息

保证这些消息最多被传送一次。对于这些消息,可靠性并非主要的考虑因素。
此模式并不要求持久性的数据存储,也不保证消息服务由于某种原因导致失败后消息不会丢失。
有两种方法指定传送模式:
1.使用setDeliveryMode 方法,这样所有的消息都采用此传送模式; 如:

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

2.使用send 方法为每一条消息设置传送模式

未经允许不得转载:菡萏如佳人 » AMQ简明教程(5)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

欢迎加入紫牛小筑

进入小筑关于作者