我正在执行一些通配符订阅(例如/a/#和/B/#)。每次订阅(参见下面的createSubscriber(topic))会产生大约1000个主题，大约需要10秒才能返回。10秒是合理的响应时间吗?在我看来它很慢，但我没有什么可以和它相比的。
给定下面的代码;
public class JMSClientSubscriber implements Runnable {
TopicConnection topicCon;
Properties properties;
List<MyListener> listeners;
JmsTopicConnectionFactory jcf;
boolean connected, alive;
public JMSClientSubscriber() throws JMSException {
properties = Properties.getInstance();
listeners = new LinkedList<>();
jcf = FLOWConnectionFactory.getTopicFactory(properties, Location.CLIENT);
connected = false;
alive = true;
}
@Override
public void run() {
try {
connect();
while (alive) {
Thread.sleep(1000);
}
disconnect();
} catch (Exception e) {
e.printStackTrace();
}
}
void connect() throws Exception {
connected = false;
topicCon = jcf.createTopicConnection();
topicCon.setExceptionListener(new ExceptionListener() {
@Override public void onException(JMSException arg0) {
disconnect();
try {
Thread.sleep(1000);
connect();
} catch (Exception e) {
e.printStackTrace();
}
}
});
topicCon.start();
for (MyListener listener: listeners) {
Thread t = new Thread() {
@Override public void run() {
TopicSession topicSes;
try {
topicSes = topicCon.createTopicSession(false, Session.DUPS_OK_ACKNOWLEDGE);
Topic topic = topicSes.createTopic(listener.exampleMessage.getTopicSubscription());
System.out.println(new Date() + " Subscribing to " + topic);
/* THIS TAKES 10 SECONDS! */ TopicSubscriber topicSub = topicSes.createSubscriber(topic);
System.out.println(new Date() + " Subscription finished " + topic);
topicSub.setMessageListener(listener);
} catch (Exception e) {
e.printStackTrace();
}
}
};
t.start();
}
connected = true;
}
void disconnect() {
try {
connected = false;
if (topicCon != null) topicCon.close();
} catch (JMSException e) {}
}
public void stop() { alive = false; }
public class MyListener implements MessageListener {
Class<? extends FlowMessage> expectedClass;
FlowMessage exampleMessage;
public MyListener(Class<? extends FlowMessage> expectedClass) throws Exception {
this.expectedClass = expectedClass;
exampleMessage = expectedClass.newInstance();
listeners.add(this);
}
@Override
public void onMessage(javax.jms.Message arg0) {
BytesMessage bm = (BytesMessage) arg0;
try {
byte bytes[] = new byte[(int) bm.getBodyLength()];
bm.readBytes(bytes);
FlowMessage flowMessage = exampleMessage.newInstance(bytes);
System.out.println(new Date() + "[" + bm.getJMSDestination() + "] " + flowMessage.toString());
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws Exception {
Properties properties = Properties.newInstance(new File("D:\cc_views\D253570_ALL_FLOW_DEV\DealingRoom\FLOW\src\cfg\flow.properties"));
LogManager.getLogManager().readConfiguration(new FileInputStream(properties.getPropertyAsFile("logging.properties")));
/* Thread per connection */
for (Class<FlowMessage> clazz: new Class[] { KondorCpty.class, KondorPair.class }) {
JMSClientSubscriber s = new JMSClientSubscriber();
s.new MyListener(clazz);
new Thread(s).start();
}
/* Thread per session */
JMSClientSubscriber s = new JMSClientSubscriber();
s.new MyListener(KondorCpty.class);
s.new MyListener(KondorPair.class);
new Thread(s).start();
}
}
此代码中的main运行两个测试;
单连接+多线程/会话
Tue Sep 13 10:18:50 2016 Subscribing to topic://DRS/OW/Cpty/#
Tue Sep 13 10:18:50 2016 Subscribing to topic://DRS/OW/Pair/#
Tue Sep 13 10:19:00 2016 Subscription finished topic://DRS/OW/Cpty/#
Tue Sep 13 10:19:07 2016 Subscription finished topic://DRS/OW/Pair/#
Tue Sep 13 10:19:08 2016[topic://DRS/OW/Pair/RONGBP] KondorPair
多线程连接+每个线程/连接一个会话
Tue Sep 13 10:22:42 2016 Subscribing to topic://DRS/OW/Pair/#
Tue Sep 13 10:22:42 2016 Subscribing to topic://DRS/OW/Cpty/#
Tue Sep 13 10:22:52 2016 Subscription finished topic://DRS/OW/Cpty/#
Tue Sep 13 10:23:00 2016 Subscription finished topic://DRS/OW/Pair/#
Tue Sep 13 10:23:00 2016[topic://DRS/OW/Pair/RONGBP] KondorPair
两个测试在时间和行为方面是相同的。
- 订阅~1000个主题需要~10秒
- 订阅似乎是顺序运行的，即使它们在不同的线程中。
- 主题更新只有在所有订阅完成后才会出现。
- 在订阅之前或之后使用TopicConnection.start()对性能或第一次Topic更新到达时没有影响。
我该如何加速呢?
请注意以下事项:
1)对于每个createSession(队列或主题)方法调用，数据从客户端和队列管理器流出，用于设置JMS会话环境。您正在使用远程连接，这意味着有数据流通过网络。
2) createSubscriber方法调用包括在队列管理器端创建订阅对象、临时队列以及主题查找、权限验证等。
你能告诉我们你是如何并行连接/会话吗?
根据JMS规范，会话不应该跨线程共享。我将为每个订阅者指定一个线程，其中线程
创建JMS连接
2)创建JMS会话3)创建订阅者
4) JMS Connection.start()是否启动消息传递
问题发生在onMessage。我没有在这里处理消息，而是将消息放在BlockingQueue上。许多独立的线程然后轮询这个BlockingQueue。这极大地提高了MessageListener的吞吐量，并将多线程问题从JMS/MQ代码中移开。