Mule jms 消费者正在慢慢挑选消息答案

【问题标题】：Mule jms consumers are picking message slowlyMule jms 消费者正在慢慢挑选消息
【发布时间】：2020-09-06 01:44:42
【问题描述】：

我正在实现 mule ESB 3.9 来使用来自活动 mq 的消息。

我有大约 15 个域组件，它实际编排到最终目的地

我已经为 jms 配置了一个连接池，并且使用 numberofconsumers 为 32，接收器中的最大线程为 500。

但我看到的是，直到每秒 4 条消息，jmeter 发送的请求都很好。但是如果我增加到每秒 5 条消息，那么我发现 jms 发送的请求平均到每秒大约 3 条消息。

关于如何配置消费者和接收者线程的数量有什么想法吗？

我的目标是每秒 10 条消息，而且我的所有 15 个组件每个最多只需要 30 毫秒来处理。所以这些组件的处理时间很好......但我看到总吞吐量有时约为 3 秒主要是因为没有立即选择消息。

请推荐

ActiveMQ xml

<destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" >
                    <!-- The constantPendingMessageLimitStrategy is used to prevent
                         slow topic consumers to block producers and affect other consumers
                         by limiting the number of messages that are retained
                         For more information, see:

                         http://activemq.apache.org/slow-consumer-handling.html

                    -->
                                                                                
                                                                                                
                  <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="10000"/>
                  </pendingMessageLimitStrategy>
                                                                  
                                                                                <slowConsumerStrategy>
                                                                                                <abortSlowConsumerStrategy/>
                                                                                  </slowConsumerStrategy>
                                                                                </policyEntry>
                                                                
                                                                
                                                                                <!--Mahesh added start-->
                                                
                                
                                                                                                <policyEntry queue=">" >
                                                                                                                  <slowConsumerStrategy>
                                                                                                                                <abortSlowConsumerStrategy/>
                                                                                                                  </slowConsumerStrategy>
                                                                                                </policyEntry>
                                
                                <!--Mahesh added end-->
                                
                                
              </policyEntries>
            </policyMap>
        </destinationPolicy>

 <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>
        
         <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="90" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="5 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="5 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>
        
           <transportConnectors>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
            <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>

Mule ESB 的其他信息

我正在使用 1 条消息的预取
消费者数量为 32，接收者线程为 500

mule-config ---->

<spring:beans>
	<context:component-scan  base-package="com.test.esb.orm.db" />
	<spring:import resource="classpath*:applicationContext.xml" />
	<spring:import resource="classpath*:test-orm-beans.xml" />
	<spring:import resource="mule-domain-spring.xml"/>
</spring:beans>

<db:generic-config name="Generic_Database_Configuration" 
					dataSource-ref="testDataSource"
					doc:name="Generic Database Configuration"/>
					
<db:generic-config name="Generic_TrackTraceDatabase_Configuration" 
					dataSource-ref="testTrackTraceDataSource"
					doc:name="Generic Database Configuration for Track Trace"/>
					
<!-- ********************* ESB Gateway Configurations ******************************** -->
<http:listener-config name="test-Shared-http-listener" host="${domain.gateway.host}" port="${domain.gateway.port}" doc:name="HTTP Listener connector for ESB Gateway (From ui)">
	<http:worker-threading-profile maxThreadsActive="64"
						poolExhaustedAction="WAIT"
						threadWaitTimeout="30000" />
</http:listener-config>

   
<jms:activemq-connector name="com.test.esb.trans.jmsConnector"
                                            username="${domain.amq.user.id}" password="${domain.amq.user.password}"
                                            connectionFactory-ref="pooledConnectionFactory"
                                            numberOfConsumers="16" acknowledgementMode="AUTO_ACKNOWLEDGE"

                                            validateConnections="true"
                                            persistentDelivery="true"
                                            doc:name="AMQ Connector For Domain Routing Services"
                                            specification="1.1">
    <receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
    <dispatcher-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
    <reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>


    <jms:activemq-connector name="com.test.esb.domain.jmsConnector" 
						username="${domain.amq.user.id}" password="${domain.amq.user.password}" 
					connectionFactory-ref="pooledConnectionFactory"
					numberOfConsumers="16" acknowledgementMode="AUTO_ACKNOWLEDGE"  
					
						validateConnections="true" 
						persistentDelivery="false" 
						doc:name="AMQ Connector For Domain Routing Services" 
						specification="1.1">
	<receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
	<dispatcher-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
	<reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>

  
<jms:activemq-connector name="com.test.esb.domain.orchestrator.jmsConnector"
                                            username="${domain.amq.user.id}" password="${domain.amq.user.password}"
                                            connectionFactory-ref="pooledConnectionFactory"
                                            numberOfConsumers="16" acknowledgementMode="AUTO_ACKNOWLEDGE"

                                            validateConnections="true"
                                            persistentDelivery="true"
                                            doc:name="AMQ Connector For Domain Routing Services"
                                            specification="1.1">
    <receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
    <dispatcher-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
    <reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>


   
<jms:activemq-connector name="com.test.esb.interface2.jmsConnector" 
						username="${domain.amq.user.id}" password="${domain.amq.user.password}" 
						connectionFactory-ref="pooledConnectionFactory"
						validateConnections="true" 
						numberOfConcurrentTransactedReceivers="16"
						persistentDelivery="false" 
						doc:name="AMQ Connector for interface2 Interface" 
						specification="1.1">
	<receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT" threadWaitTimeout="30000"/>
	<dispatcher-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
	<reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>

   
<jms:activemq-connector name="com.test.esb.destination.router.jmsConnector" 
						username="${domain.amq.user.id}" password="${domain.amq.user.password}" 
						connectionFactory-ref="pooledConnectionFactory"
						validateConnections="true" 
						numberOfConcurrentTransactedReceivers="8"
						persistentDelivery="false" 
						doc:name="AMQ Connector for DestinationRouter" 
						specification="1.1">
	<receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
	<dispatcher-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT"/>
	<reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>

   
<jms:activemq-connector name="com.test.esb.dlq.jmsConnector" 
						username="${domain.amq.user.id}" password="${domain.amq.user.password}" 
						connectionFactory-ref="pooledConnectionFactory"
						validateConnections="true" 
						numberOfConcurrentTransactedReceivers="16"
						persistentDelivery="false" 
						doc:name="AMQ Connector for Dead Letter Queue" 
						specification="1.1">
	<receiver-threading-profile maxThreadsActive="500" poolExhaustedAction="WAIT" threadWaitTimeout="30000"/>
	<reconnect count="15" frequency="5000" blocking="true"/>
</jms:activemq-connector>

mule-spring 配置---->

<spring:bean id="domainRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
		<spring:property name="maximumRedeliveries" value="5" />
		<spring:property name="initialRedeliveryDelay" value="500" />
		<spring:property name="maximumRedeliveryDelay" value="10000" />
		<spring:property name="useExponentialBackOff" value="false" />
		<spring:property name="backOffMultiplier" value="3" />
	</spring:bean>
	
	
	

	
	<!-- ActiveMQ Connection factory -->
	<spring:bean id="domainConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
			value="tcp://192.0.0.0:61616?jms.prefetchPolicy.all=1" />
		<spring:property name="redeliveryPolicy" ref="domainRedeliveryPolicy" />
	</spring:bean>
	
	
		<!-- amqExceptionConnectionFactory Connection factory -->
	<spring:bean id="amqExceptionConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
			value="failover:(${domain.amq.failover.url}?jms.prefetchPolicy.all=50)" />
		<spring:property name="redeliveryPolicy" ref="domainRedeliveryPolicy" />
	</spring:bean>
	
	
	<spring:bean id="pooledConnectionFactory" class="org.apache.activemq.jms.pool.PooledConnectionFactory" >
		<spring:property name="connectionFactory" ref="domainConnectionFactory"/>
        	<spring:property name="maxConnections" value="10000" />
        	<spring:property name="maximumActiveSessionPerConnection" value="10000" />
    	</spring:bean>

	<spring:bean id="trackTraceRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
		<spring:property name="maximumRedeliveries" value="5" />
		<spring:property name="initialRedeliveryDelay" value="500" />
		<spring:property name="maximumRedeliveryDelay" value="10000" />
		<spring:property name="useExponentialBackOff" value="false" />
		<spring:property name="backOffMultiplier" value="3" />
	</spring:bean>


	<spring:bean id="trackTraceConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
		value="failover:(${domain.amq.failover.url})" />
			<!-- value="tcp://${domain.amq.host}:${domain.amq.port}" /> -->
			
		<spring:property name="redeliveryPolicy" ref="trackTraceRedeliveryPolicy" />
	</spring:bean>

	

	<spring:bean id="interface2ConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
		value="tcp://192.0.0.0:61616?jms.prefetchPolicy.all=1" />
			<!-- value="tcp://${domain.amq.host}:${domain.amq.port}" /> -->
		<spring:property name="redeliveryPolicy" ref="interface2RedeliveryPolicy" />
	</spring:bean>

	<spring:bean id="interface2RedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
		<spring:property name="maximumRedeliveries" value="5" />
		<spring:property name="initialRedeliveryDelay" value="500" />
		<spring:property name="maximumRedeliveryDelay" value="10000" />
		<spring:property name="useExponentialBackOff" value="false" />
		<spring:property name="backOffMultiplier" value="3" />
	</spring:bean>

	<!-- ********************* Destination Router Configurations ******************************** -->
	<!-- ActiveMQ Connection factory for destination Router -->
	<spring:bean id="destinationRouterConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
		value="tcp://192.0.0.0:61616?jms.prefetchPolicy.all=1" />
			<!-- value="tcp://${domain.amq.host}:${domain.amq.port}" /> -->
		<spring:property name="redeliveryPolicy"
			ref="destinationRouterRedeliveryPolicy" />
	</spring:bean>

	<spring:bean id="destinationRouterRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
		<spring:property name="maximumRedeliveries" value="5" />
		<spring:property name="initialRedeliveryDelay" value="500" />
		<spring:property name="maximumRedeliveryDelay" value="10000" />
		<spring:property name="useExponentialBackOff" value="false" />
		<spring:property name="backOffMultiplier" value="3" />
	</spring:bean>

	<!-- ********************* Dead Letter Queue (DLQ) Configurations ******************************** -->
	<!-- ActiveMQ Connection factory for Dead Letter Queue -->
	<spring:bean id="dlqConnectionFactory"
		class="org.apache.activemq.ActiveMQConnectionFactory" lazy-init="true">
		<spring:property name="brokerURL"
		value="failover:(${domain.amq.failover.url})" />
			
		<spring:property name="redeliveryPolicy" ref="dlqRedeliveryPolicy" />
	</spring:bean>

	<spring:bean id="dlqRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
		<spring:property name="maximumRedeliveries" value="5" />
		<spring:property name="initialRedeliveryDelay" value="500" />
		<spring:property name="maximumRedeliveryDelay" value="10000" />
		<spring:property name="useExponentialBackOff" value="false" />
		<spring:property name="backOffMultiplier" value="3" />
	</spring:bean>

【问题讨论】：

可能有一些东西要添加到activemq.xml——你的activemq.xml配置是什么？
@MattPavlovich 请在上面找到我对 activemq.xml 详细信息的编辑问题
太好了，您所拥有的大部分内容都使用默认值，这很容易超出您的目标性能数字。您可以尝试添加 queuePreetch="128" 和优化的Dispatch="true"，但我怀疑问题是客户端连接工厂我们的 Mule 效率低下。许多像 Spring JMSTemplate 这样的框架都有很高的延迟，这使得消耗看起来很慢。我会写一个快速的 JMS 消费者来确认它不是代理（我怀疑它不是代理）。
您可以在 JMS 端点上发布连接工厂设置和任何 mule 配置吗？
@MattPavlovich，我已更新供您参考。您的专业知识可能会有所帮助

标签： activemq mule-component mule-esb

【解决方案1】：

numberOfConsumers 是确定端点使用事务时客户端线程数的属性。如果端点处于事务中，则您需要改用numberOfConcurrentTransactedReceivers。

您确实应该删除 <dispatcher-threading-profile> 和 <receiver-threading-profile> 配置。每个 JMS 配置 500 个线程对您的系统来说已经很多了，如果创建它们，这将是很多资源，并且对于这个用例根本没有帮助。您应该仅出于某些真正原因设置这些值。

在处理缓慢时尝试捕获线程转储以分析线程在做什么。这应该指向根本原因。

参考：https://help.mulesoft.com/s/article/JMS-Usage-of-numberOfConsumers-and-numberOfConcurrentTransactedReceivers

【讨论】：

mule 是否有可能识别它是 numberofconsumers 还是 numberoftransactedrecievers？因为我看到即使我指定了 numberofconsumers，它也会在 mule 域日志中显示 numberoftransactedrecievers-