java慢速MQ主题订阅。并行化不能提高性能

Question

我正在执行一些通配符订阅（例如/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秒
• 订阅似乎是按顺序运行的，即使它们位于不同的线程中
• 主题更新仅在所有订阅完成后显示
• 有话题联系的。订阅之前或之后的start（）在性能或第一个主题更新到达时没有区别

那么我该如何加速呢

Answer 1

问题出在{}。我没有在这里处理消息，而是将消息放在BlockingQueue上。然后，许多独立的线程轮询了这个BlockingQueue。这极大地提高了MessageListener的吞吐量，并将多线程问题从JMS/MQ代码中移开

Answer 2

请注意以下几点：

1）对于每个createSession（队列或主题）方法调用，数据流都来自客户机和队列管理器，用于设置JMS会话环境。您使用的是远程连接，这意味着涉及网络上的数据流

2）createSubscriber方法调用涉及创建订阅对象、主题查找之外的临时队列、队列管理器端的权限验证等

你能告诉我们你是如何并行连接/会话的吗

根据JMS规范，会话不应跨线程共享。我会为每个订阅用户分配一个线程

1）创建JMS连接

2）创建JMS会话

3）创建订阅服务器

4）进行JMS连接。start（）启动消息传递