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

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

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

布鲁斯

Answer 2

尽管这是一个老问题，但缺少一个非常非常重要的建议：

• 调查您拥有的主题和队列的数量。

ActiveMQ 将订阅主题保留在单独的线程中。特别是，当您有大量不同的主题时，这会拖累任何服务器。考虑使用 JMS 选择器。

我遇到了类似的情况，每秒有数千条市场数据消息。当我天真地将每条消息转储到市场工具特定通道中时，服务器能够在向消息生产者吐出错误消息之前等待大约一个小时。我更改了设计以拥有一个通道“MARKET_DATA”，然后在所有生成的消息上设置标头属性，并在消费者端设置一个选择器以仅选择我想要的消息。请注意，我的选择器采用类似 SQL 的语法，并且在服务器上运行...（是的，让我们跳过 CEP 营销炒作）...

Answer 3

我们不使用ActiveMQ，但我们遇到了类似的问题，我们发现问题出在后端处理而不是Java端。这里可能存在多个问题：

1. 处理来自队列的消息的程序可能很慢（例如大型机上的 CICS），它可能无法跟上发送到队列的消息。一种可能的解决方案是提高处理能力（或优化处理消息的后端代码）
2. 检查队列上的消息，有时队列上有很多未提交的有害消息，我们使用单独的队列来处理此类消息消息。