WLS_078：WebLogic Server高级管理之七：JMS Topic 持久化订阅

运行环境：WebLogic Server 12.1.1 开发版 + Oracle Database 10g Express Edition 10.2.0.1。

默认情况下，如果Topic的订阅者“不在线”，那么是无法收到Topic中的消息的。
这与一些实际情况不相符合：我订阅的杂志，我不在家，就不给我了吗？
为此，WebLogic Server提供了一个Topic持久化订阅功能。
它可以帮助我们在“离线”的情况下，上线后收到之前所有没有收到的消息。

重要步骤说明：
1. 实验准备：创建和配置JMS资源
请参考实验《为JMS消息设置过期时间》。

2. 创建File Store
Topic中的消息要想做到持久化，必须设定持久化方式，这里我们使用File Store作为Persistent Store。


3. 为JMS Server设置Persistent Store


4. 为Topic 创建Durable Subscriber
点击Topic，选择Monitor-->Durable Subscribers-->New
设置Subscription Name：ChinaSubscripiton
设置Client ID：ChinaClient


单凭Client ID还不足以唯一标识一个持久化订阅者，因为一个Client ID还可以该Topic中的多种“杂志”。
Subscription Name就相当于杂志的名称。这跟一个身份证可以订多个房间的道理类似。

5. 编写Topic 发送方代码：TopicSend.java
public void init(Context ctx, String topicName) throws NamingException, JMSException {
tconFactory = (TopicConnectionFactory)ctx.lookup(JMS_FACTORY);
tcon = tconFactory.createTopicConnection();
tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (Topic)ctx.lookup(topicName);
tpublisher = tsession.createPublisher(topic);
msg = tsession.createTextMessage();
tcon.start();
}

private static void readAndSend(TopicSend ts) throws IOException, JMSException {
BufferedReader msgStream = new BufferedReader(new InputStreamReader(System.in));
String line = null;
boolean quitNow = false;
do {
System.out.print("Enter message (\"quit\" to quit): \n");
line = msgStream.readLine();
if (line != null && line.trim().length() != 0) {
ts.send(line);
System.out.println("JMS Message Sent: " + line + "\n");
quitNow = line.equalsIgnoreCase("quit");
}
} while (!quitNow);
}
以上代码跟发送到普通的Topic代码没有什么不同。
区别是在Topic接收方代码。

6. 编写普通的Topic 接收方代码：TopicReceiveClient.java
public void init(Context ctx, String topicName) throws NamingException, JMSException {
tconFactory = (TopicConnectionFactory)ctx.lookup(JMS_FACTORY);
tcon = tconFactory.createTopicConnection();
tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (Topic)ctx.lookup(topicName);
tsubscriber = tsession.createSubscriber(topic);
tsubscriber.setMessageListener(this);
tcon.start();
}

public void onMessage(Message msg) {
try {
String msgText;
if (msg instanceof TextMessage) {
msgText = ((TextMessage)msg).getText();
} else {
msgText = msg.toString();
}

System.out.println("Message Received: " + msgText);

if (msgText.equalsIgnoreCase("quit")) {
synchronized (this) {
quit = true;
this.notifyAll(); // Notify main thread to quit
}
}
} catch (JMSException jmse) {
System.err.println("An exception occurred: " + jmse.getMessage());
}
}

7. 编写持久化的Topic 接收方代码：DurableTopicReceiveClient.java
public void init(Context ctx, String topicName) throws NamingException, JMSException {
tconFactory = (TopicConnectionFactory)ctx.lookup(JMS_FACTORY);
tcon = tconFactory.createTopicConnection();
tcon.setClientID("ChinaClient");
tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (Topic)ctx.lookup(topicName);
tsubscriber = tsession.createDurableSubscriber(topic,"ChinaSubscription");
tsubscriber.setMessageListener(this);
tcon.start();
}

onMessage方法与普通的Topic 接收方代码相同。

8. 测试

8.1 测试环境设置
打开4个DOS命令窗口，均设置如下环境变量
set JAVA_HOME=C:\Oracle\Java\jdk1.6.0_25
set MW_HOME=C:\Oracle\wls1211_dev
set JAVA_VENDOR=Sun
运行脚本：%MW_HOME%\wlserver\server\bin\setWLSEnv.cmd

8.2 测试普通Topic订阅者特性：一对多的消息分发
（1）在DOS命令窗口中启动AdminServer。
（2）在DOS命令窗口中启动消息生产者：java TopicSend t3://localhost:7001
（3）在两个DOS命令窗口中分别启动两个消息消费者：java TopicReceiveClient t3://localhost:7001
（4）在消息生产者DOS窗口连续发送4条消息：msg1、msg2、msg3、msg4。
（5）观察两个消息消费者DOS窗口，发现两个窗口都“消费”了所有消息。
（6）在消息生产者DOS命令窗口输入quit，两个消息消费者DOS窗口都收到该消息，并退出。

8.3 测试普通Topic订阅者特性：离线后再上线，无法收到以前的消息
（1）在DOS命令窗口中启动AdminServer。
（2）在DOS命令窗口中启动消息生产者：java TopicSend t3://localhost:7001
（3）在1个DOS命令窗口中启动1个消息消费者：java TopicReceiveClient t3://localhost:7001
（4）在消息生产者DOS窗口连续发送2条消息：msg1、msg2。
（5）在1个DOS命令窗口中启动另外1个消息消费者：java TopicReceiveClient t3://localhost:7001
（6）在消息生产者DOS窗口连续发送2条消息：msg3、msg4。
（7）观察两个消息消费者DOS窗口，发现第1个窗口“消费”了所有消息，第2个窗口只“消费”了msg3、msg4。
（8）在消息生产者DOS命令窗口输入quit，两个消息消费者DOS窗口都收到该消息，并退出。

8.4 测试持久化Topic订阅者特性：离线后再上线，依然可以收到以前的消息
（1）在DOS命令窗口中启动AdminServer。
（2）在DOS命令窗口中启动消息生产者：java TopicSend t3://localhost:7001
（3）在1个DOS命令窗口中启动1个消息消费者：java TopicReceiveClient t3://localhost:7001
（4）在消息生产者DOS窗口连续发送2条消息：msg1、msg2。
（5）在1个DOS命令窗口中启动另外1个持久化消息消费者：java DurableTopicReceiveClient t3://localhost:7001
（6）在消息生产者DOS窗口连续发送2条消息：msg3、msg4。
（7）观察两个消息消费者DOS窗口，发现两个窗口都“消费”了所有消息，这说明持久化配置起作用了。
（8）在消息生产者DOS命令窗口输入quit，两个消息消费者DOS窗口都收到该消息，并退出。

8.5 测试持久化Topic订阅者特性：设置多个Subscription Name
（1）在DOS命令窗口中启动AdminServer。
（2）在DOS命令窗口中启动消息生产者：java TopicSend t3://localhost:7001
（3）在两个DOS命令窗口中分别启动两个消息消费者：java DurableTopicReceiveClient t3://localhost:7001
这时，在第2个消费者窗口会报错：说Client id:ChinaClient已经被另一个对象使用了。
Exception in thread "main" weblogic.jms.common.InvalidClientIDException: Client id, ChinaClient, is in use. The reason for rejection is "The JNDI name weblogic
.jms.connection.clientid.ChinaClient was found, and was bound to an object of type weblogic.jms.frontend.FEClientIDSingularAggregatable : FEClientIDSingularAggr
egatable(SingularAggregatable(<6539122430256738311 .1="">:4):ChinaClient)"
at weblogic.jms.dispatcher.DispatcherAdapter.convertToJMSExceptionAndThrow(DispatcherAdapter.java:110)
at weblogic.jms.dispatcher.DispatcherAdapter.dispatchSync(DispatcherAdapter.java:45)
at weblogic.jms.client.JMSConnection.setClientIDInternal(JMSConnection.java:660)
at weblogic.jms.client.JMSConnection.setClientID(JMSConnection.java:630)
at weblogic.jms.client.WLConnectionImpl.setClientID(WLConnectionImpl.java:662)
at DurableTopicReceiveClient.init(DurableTopicReceiveClient.java:72)
at DurableTopicReceiveClient.main(DurableTopicReceiveClient.java:104)
Caused by: weblogic.jms.common.InvalidClientIDException: Client id, ChinaClient, is in use. The reason for rejection is "The JNDI name weblogic.jms.connection.
clientid.ChinaClient was found, and was bound to an object of type weblogic.jms.frontend.FEClientIDSingularAggregatable : FEClientIDSingularAggregatable(Singula
rAggregatable(<6539122430256738311 .1="">:4):ChinaClient)"
at weblogic.jms.frontend.FEConnection.setConnectionClientId(FEConnection.java:1170)
at weblogic.jms.frontend.FEConnection.invoke(FEConnection.java:1649)
at weblogic.messaging.dispatcher.Request.wrappedFiniteStateMachine(Request.java:961)
at weblogic.messaging.dispatcher.DispatcherServerRef.invoke(DispatcherServerRef.java:276)
at weblogic.messaging.dispatcher.DispatcherServerRef.handleRequest(DispatcherServerRef.java:141)
at weblogic.messaging.dispatcher.DispatcherServerRef.access$000(DispatcherServerRef.java:34)
at weblogic.messaging.dispatcher.DispatcherServerRef$2.run(DispatcherServerRef.java:111)
at weblogic.work.ExecuteThread.execute(ExecuteThread.java:256)
at weblogic.work.ExecuteThread.run(ExecuteThread.java:221)

这说明，Client ID只能被一个持久化Topic接收方使用，具有排他性。也就是说，多个持久化Topic接收方，需要使用多个Client ID。
（4）复制DurableTopicReceiveClient.java，生成DurableTopicReceiveClient2.java
只把China的地方改成Beijing。
public void init(Context ctx, String topicName) throws NamingException, JMSException {
tconFactory = (TopicConnectionFactory)ctx.lookup(JMS_FACTORY);
tcon = tconFactory.createTopicConnection();
tcon.setClientID("BeijingClient");
tsession = tcon.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
topic = (Topic)ctx.lookup(topicName);
tsubscriber = tsession.createDurableSubscriber(topic,"BeijingSubscription");
tsubscriber.setMessageListener(this);
tcon.start();
}
（5）在1个DOS命令窗口中启动消息消费者：java DurableTopicReceiveClient2 t3://localhost:7001
我们意外地发现，在Console中的Topic Durable Subscribers中“自动”增加了Beijing持久化订阅者。
也就是说，不用在Console事先配置，这个特性比较好，省得为每一个持久化订阅者人工配置。


（6）在消息生产者DOS窗口连续发送2条消息：msg1、msg2。
（7）观察两个消息消费者DOS窗口，发现两个窗口都“消费”了所有消息。
（8）停掉两个持久化消费者。
（9）在消息生产者DOS窗口连续发送2条消息：msg3、msg4。
（10）观察两个消息消费者DOS窗口，发现两个窗口都“消费”了所有消息。

代码下载：DurableTopicSubscriber.7z