一、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等)
三、topic
3.1 关于持久化和非持久化消息
持久化消息:
这是ActiveMQ的默认传送模式,此模式保证这些消息只被传送一次和成功使用一次。对于这些消息,可靠性是优先考虑的因素。
可靠性的另一个重要方面是确保持久性消息传送至目标后,消息服务在向消费者传送它们之前不会丢失这些消息。
这意味着在持久性消息传送至目标时,消息服务将其放入持久性数据存储。如果消息服务由于某种原因导致失败,它可以恢复
此消息并将此消息传送至相应的消费者。虽然这样增加了消息传送的开销,但却增加了可靠性。
非持久化消息:
保证这些消息最多被传送一次。对于这些消息,可靠性并非主要的考虑因素。 此模式并不要求持久性的数据存储,也不保证
消息服务由于某种原因导致失败后消息不会丢失。 有两种方法指定传送模式:
1.使用setDeliveryMode 方法,这样所有的消息都采用此传送模式; 如:
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
2.使用send 方法为每一条消息设置传送模式
3.2 非持久的Topic消息
3.2.1 消息的发送
基本跟前面发送队列信息是一样的,只是把创建Destination的地方,由创建队列替换成创建Topic,例如:
Destination destination = session.createTopic("my-topic");
3.2.1 消息的接收
1:必须要接收方在线,然后客户端再发送信息,接收方才能接收到消息
2:同样把创建Destination的地方,由创建队列替换成创建Topic,例如:
Destination destination = session.createTopic("my-topic");
3:由于不知道客户端发送多少信息,因此改成while循环的方式了,例如:
Message message = consumer.receive(); while(message!=null) { TextMessage txtMsg = (TextMessage)message; System.out.println("收到消 息:" + txtMsg.getText()); message = consumer.receive(1000L); }
3.3 持久的Topic消息
3.3.1 消息的发送
/**持久的*/ public void test2() throws Exception { ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.91.8: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); // 设置DeliveryMode.PERSISTENT模式 connection.start(); for (int i = 0; i < 3; i++) { TextMessage message = session.createTextMessage("message22--" + i); producer.send(message); } session.commit(); session.close(); connection.close(); }
1:要用持久化订阅,发送消息者要用DeliveryMode.PERSISTENT模式发现,在连接之前设定
2:一定要设置完成后,再start这个connection
3.3.2 消息的接收
public void test2() throws Exception { ConnectionFactory cf = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.91.8:61616"); Connection connection = cf.createConnection(); connection.setClientID("cc1"); final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Topic topic = session.createTopic("my-topic"); TopicSubscriber ts = session.createDurableSubscriber(topic, "T1"); connection.start(); Message message = ts.receive(); while(message!=null) { TextMessage txtMsg = (TextMessage)message; System.out.println("收到消息:" + txtMsg.getText()); message = ts.receive(1000L); } session.commit(); session.close(); connection.close(); }
1:需要在连接上设置消费者id,用来识别消费者
2:需要创建TopicSubscriber来订阅
3:要设置好了过后再start 这个 connection
4:一定要先运行一次,等于向消息服务中间件注册这个消费者,然后再运行客户端发送信息,这个时候,无论消费者是否在线,
都会接收到,不在线的话,下次连接的时候,会把没有收过的消息都接收下来。