java - JMS (ActiveMQ) 性能

Question

我有一个 Java 应用程序，其中有许多组件通过 JMS (ActiveMQ) 进行通信。目前，应用程序和 JMS Hub 位于同一服务器上，尽管我们最终计划拆分组件以实现可扩展性。我们在性能方面存在重大问题，最值得注意的是，所有这些问题似乎都与 JMS 相关，而这个问题的焦点是向主题发布消息所花费的时间。

我们有大约 50 个动态创建的主题，用于应用程序组件之间的通信。一个组件从表中读取记录并一次处理一条记录，该处理涉及创建一条 JMS 对象消息并将其发布到其中一个主题。此处理无法跟上将记录写入源表的速率~23/秒，因此我们更改了处理以创建 JMS 对象消息并将其添加到队列中。创建了一个新线程，该线程从此队列中读取并将消息发布到适当的主题。显然，这不会加快处理速度，但它确实使我们能够通过查看队列的大小来了解我们落后了多远。

当天开始时，整个系统没有消息通过，这会迅速从第一个小时内通过集线器的 1560000 条(433 条/秒)消息增加到第三个小时内的 2100000 条(582 条/秒)消息，然后保持在那个水平。在第一个小时开始时，从数据库表中读取记录的组件发布的消息保持不变，但是到该小时结束时，队列中有 2000 条消息等待发送，到了第 3 小时，队列中有 9000 条消息其中的消息。

下面是发送 JMS 消息的代码的相应部分。非常感谢任何有关我们做错了什么或如何提高性能的建议。查看网络上的统计数据，JMS 应该能够轻松处理每秒约 1000-2000 条大消息或每秒约 10000 条小消息。我们的消息每条大约 500 字节，所以我想它位于该范围的中间位置。

获取发布者的代码:

private JmsSessionPublisher getJmsSessionPublisher(String topicName) throws JMSException {
    if (!this.topicPublishers.containsKey(topicName)) {
        TopicSession pubSession = (ActiveMQTopicSession) topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

        ActiveMQTopic topic = getTopic(topicName, pubSession);

        // Create a JMS publisher and subscriber
        TopicPublisher publisher = pubSession.createPublisher(topic);

        this.topicPublishers.put(topicName, new JmsSessionPublisher(pubSession, publisher));
    }
    return this.topicPublishers.get(topicName);
}

发送消息:

JmsSessionPublisher jmsSessionPublisher = getJmsSessionPublisher(topicName);
    
ObjectMessage objMessage = jmsSessionPublisher.getSession().createObjectMessage(messageObj);
objMessage.setJMSCorrelationID(correlationID);
objMessage.setJMSTimestamp(System.currentTimeMillis());
jmsSessionPublisher.getPublisher().publish(objMessage, false, 4, 0);

将消息添加到队列的代码:

List<EventQueue> events = eventQueueDao.getNonProcessedEvents();
for (EventQueue eventRow : events) {
    IEvent event = eventRow.getEvent();
    AbstractEventFactory.EventType eventType = AbstractEventFactory.EventType.valueOf(event.getEventType());
    String topic = event.getTopicName() + topicSuffix;
    EventMsgPayload eventMsg = AbstractEventFactory.getFactory(eventType).getEventMsgPayload(event);
    synchronized (queue) {
        queue.add(new QueueElement(eventRow.getEventId(), topic, eventMsg));
        queue.notify();
    }
}

线程中的代码从队列中删除项目:

jmsSessionFactory.publishMessageToTopic(e.getTopic(), e.getEventMsg(), Integer.toString(e.getEventMsg().hashCode()));

publishMessageToTopic 执行上面的“发送消息”代码。

如果一致认为 ActiveMQ 可能不是最佳选择，则可以选择其他 JMS 实现。

Answer 1

并不是 100% 清楚您在哪里遇到性能缓慢的情况，但听起来您所描述的是发布消息的速度缓慢。您每次发布消息时都会创建一个新的发布者吗？如果是这样，那么效率非常低，您应该考虑创建一个发布者并一遍又一遍地使用它来发送消息。此外，如果您要发送持久消息，那么您可能正在使用同步发送到代理。您可能需要考虑使用异步发送来加快速度。有关更多信息，请参阅有关 Async Sends 的文档

另外，消费者的表现如何？有多少消费者正在使用？他们能否跟上消息发布的速度？

此外，您正在使用的代理配置是什么？它已经被调整了吗？

布鲁斯