高性能默认消息侦听器容器配置，适用于数千个主题

对于我当前的项目，我需要使用来自许多目的地的消息（从数百个到20或30k）所有目的地都是主题。目前（对于初始负载测试）所有消息都在同一台服务器上的线程池中本地创建。

我当前的 Spring 配置在代理网络（用于集群）和 DefaultMessageListenerContainers （DMLC）中使用嵌入式 ActiveMQ，并带有一个通用的 TaskExecutor。虽然目的地的数量非常高，但每个目的地的吞吐量相对较低。

我唯一的要求是尽快使用所有消息。

我的配置：

<bean id="connectionfactory" class="org.springframework.jms.connection.CachingConnectionFactory" destroy-method="destroy">
    <property name="targetConnectionFactory">
        <ref bean="amqConnectionFactory" />
    </property>
</bean>
<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="vm://localhost:61616?async=false&amp;jms.dispatchAsync=false" />
    <property name="userName" value="admin" />
    <property name="password" value="admin" />
</bean>
<bean id="listenerThreadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="70" />
    <property name="maxPoolSize" value="70" />
    <property name="daemon" value="true" />
    <property name="keepAliveSeconds" value="60" />
</bean>
<!-- Message Listener Container Template for Topics -->
<bean id="topiccontainertemplate" class="org.springframework.jms.listener.DefaultMessageListenerContainer" scope="prototype"
    destroy-method="destroy">
    <property name="autoStartup" value="false" />
    <property name="connectionFactory" ref="connectionfactory" />
    <property name="pubSubDomain" value="true" />
    <property name="cacheLevelName" value="CACHE_CONSUMER" />
    <property name="destinationName" value="default" />
    <property name="maxMessagesPerTask" value="1" />
    <property name="receiveTimeout" value="1" />
    <property name="taskExecutor" ref="listenerThreadPoolTaskExecutor" />
</bean>

我的代码将应用程序上下文用作 DMLC 工厂，并设置容器的最终配置：

    AbstractMessageListenerContainer container = context.getBean("simpletopiccontainertemplate", AbstractMessageListenerContainer.class);
    container.setDestinationName(localEntity.getId().getDestination());
    container.setMessageListener(mylistener);
    container.start();

虽然我们不会在此配置中丢失消息，但单个消息的转换时间可能很长。

Q1：有没有更有效的方式来收听大量目的地？

问题 2：我的侦听器配置是否有可能的改进？

Q3：除了DMLC之外，我还尝试了SimpleMessageListenerContainer，但我无法让它工作。我的配置有问题吗？

<bean id="simpletopiccontainertemplate" class="org.springframework.jms.listener.SimpleMessageListenerContainer" scope="prototype"
    destroy-method="destroy">
    <property name="autoStartup" value="false" />
    <property name="connectionFactory" ref="connectionfactory" />
    <property name="destinationName" value="default" />
    <property name="concurrency" value="1" />
    <property name="pubSubDomain" value="true" />
    <property name="taskExecutor" ref="listenerThreadPoolTaskExecutor" />
</bean>

我是否正确理解您正在手动创建 20-30 千个SimpleMessageListenerContainer，每个都在听不同的主题？我很惊讶它甚至可以工作，因为默认情况下每个侦听器都会创建一个线程，并且在一个 JVM 中同时运行的 20-30 个数千个线程令人印象深刻（也很可怕）。

我看到两种方法可以提高您的表现：

手动阅读消息，例如使用JmsTemplate.receive()方便的方法。只要记住使用非常低的超时。通过在更少的线程（10？100？）中迭代检查每个主题，您将遇到一些延迟，但更少的线程意味着更少的上下文切换时间。

老实说，我不认为这个解决方案会扩展并为您提供更好的性能，但请尝试检查不同数量的线程和超时。
簇！恐怕用一个应用程序收听 20-30k 个主题永远不会奏效（足够快）。但是，似乎可以轻松横向扩展此任务。投入十台或一百台服务器，让每台服务器相应地监听两千或两百个主题。这将线性扩展。

我从DefaultMessageListenerContainer搬到了SimpleMessageListenerContainer。简单容器使用推送消息传递。这样（并且通过适当的配置，嵌入式代理将消息推送到侦听器中，而不是轮询目标。

相关内容

最新更新

热门标签：