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

AMQ简明教程（5） JMS的API开发步骤

2018-01-30 分类：架构类 阅读(2985) 评论(0)

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 方法为每一条消息设置传送模式

AD：【极客江湖】
未经允许不得转载：菡萏如佳人 » AMQ简明教程（5）
标签：

相关推荐

欢迎加入极客江湖

进入江湖关于作者