一、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 方法,这样所有的消息都采用此传送模式; 如:
1
|
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
|
2.使用send 方法为每一条消息设置传送模式
3.2 非持久的Topic消息
3.2.1 消息的发送
基本跟前面发送队列信息是一样的,只是把创建Destination的地方,由创建队列替换成创建Topic,例如:
1
|
Destination destination = session.createTopic(
"my-topic"
);
|
3.2.1 消息的接收
1:必须要接收方在线,然后客户端再发送信息,接收方才能接收到消息
2:同样把创建Destination的地方,由创建队列替换成创建Topic,例如:
1
|
Destination destination = session.createTopic(
"my-topic"
);
|
3:由于不知道客户端发送多少信息,因此改成while循环的方式了,例如:
1
2
3
4
5
6
|
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 消息的发送
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
/**持久的*/
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 消息的接收
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
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:一定要先运行一次,等于向消息服务中间件注册这个消费者,然后再运行客户端发送信息,这个时候,无论消费者是否在线,
都会接收到,不在线的话,下次连接的时候,会把没有收过的消息都接收下来。
本文转自我爱大金子博客51CTO博客,原文链接http://blog.51cto.com/1754966750/1914965如需转载请自行联系原作者
我爱大金子