具有Jms连接器的ule子流，动态出站端点中的线程阻塞 [英] Mule flow with Jms connector, Threads blocking in dynamic outbound endpoint

问题描述

我有一个jms连接器，我正在从队列中接收消息，以处理流中的消息，调用db以基于消息中的某些ID获取数据并将响应输出写入文件，我正在使用动态出站端点来确定输出位置.

I have a jms connector, i am receiving message from a queue processing the message in a flow, calling db to get the data based on some ids in the message and writing response output to files, i am using dynamic outbound endpoints to decide output location.

    <jms:connector name="tibco" numberOfConsumers="20" ..... >
                  .....
    </jms:connector>
    <flow name="realtime" doc:name="ServiceId-8">
    <jms:inbound-endpoint queue="${some.queue}" connector-ref="tibco" doc:name="JMS">
       <jms:transaction action="ALWAYS_BEGIN"/>
    </jms:inbound-endpoint>
    <processor ref="proc1"></processor>
    <processor ref="proc2"></processor>
    <component doc:name="Java">
        <spring-object bean="comp1"/>
    </component>
    <processor ref="proc3"></processor>
    <collection-splitter doc:name="Collection Splitter"/>
    <processor ref="endpointprocessor"></processor>
    <foreach collection="#[message.payload.consumerEndpoints]" counterVariableName="endpoints" doc:name="Foreach">

            <when expression="#[consumerEndpoint.getOutputType().equals('txt') and consumerEndpoint.getChannel().equals('file')]">
                <processor-chain>
                    <file:outbound-endpoint path="#[consumerEndpoint.getPath()]" outputPattern="#[consumerEndpoint.getClientId()]-#[attributes['eventId']]%#[consumerEndpoint.getTicSeedCount()]-#[attributes['dateTime']].tic" responseTimeout="10000" doc:name="File"/>
                </processor-chain>  
            </when>
            <when expression="#[consumerEndpoint.getOutputType().equals('txt') and consumerEndpoint.getChannel().equals('ftp')]">
                 <processor-chain>
                    <ftp:outbound-endpoint path="#[consumerEndpoint.getPath()]" outputPattern="#[consumerEndpoint.getClientId()]-#[attributes['eventId']]%#[consumerEndpoint.getTicSeedCount()]-#[attributes['dateTime']].tic" host="#[consumerEndpoint.getHost()]" port="#[consumerEndpoint.getPort()]" user="#[consumerEndpoint.getChannelUser()]" password="#[consumerEndpoint.getChannelPass()]" responseTimeout="10000" doc:name="FTP"/>
                 </processor-chain>
            </when>
         </choice>
    </foreach>
    <rollback-exception-strategy doc:name="Rollback Exception Strategy">
        <processor ref="catchExceptionCustomHandling"></processor>
    </rollback-exception-strategy>
</flow>

以上未完成流程.我粘贴了要理解的重要部分.

Above is not complete flow. i pasted the important parts to understand.

问题1.由于我尚未在任何级别定义任何线程策略，并且连接器具有numberOfConsumers ="20"，因此，如果我在队列中丢弃20条消息，将启动多少个线程. jms队列中的预取大小设置为20.

Question 1. As i have not defined any thread strategy at any level, and connector has numberOfConsumers="20", if i drop 20 messages in queue how many threads will start. prefetch size in the jms queue is set to 20.

问题2:我是否需要在接收器端和/或流级别配置线程策略.

Question 2: Do i need to configure threading strategy at receiver end and/or at flow level.

有时负载非常高(假设一分钟内队列中有15k msgs)，我看到消息处理变慢并且线程转储显示如下内容:

some time when the load is very high(let say 15k msgs in queue in a minute) i see message processing gets slow and thread dump shows some thing like below:

"TIBCO EMS会话分派器(7905958)" prio = 10 tid = 0x00002aaadd4cf000 nid = 0x3714等待监视器条目[0x000000004af1e000] java.lang.Thread.State:已阻止(在对象监视器上) 在org.mule.endpoint.DynamicOutboundEndpoint.createStaticEndpoint(DynamicOutboundEndpoint.java:153) -等待锁定< 0x00002aaab711c0e0>(一个org.mule.endpoint.DynamicOutboundEndpoint)

"TIBCO EMS Session Dispatcher (7905958)" prio=10 tid=0x00002aaadd4cf000 nid=0x3714 waiting for monitor entry [0x000000004af1e000] java.lang.Thread.State: BLOCKED (on object monitor) at org.mule.endpoint.DynamicOutboundEndpoint.createStaticEndpoint(DynamicOutboundEndpoint.java:153) - waiting to lock <0x00002aaab711c0e0> (a org.mule.endpoint.DynamicOutboundEndpoint)

任何帮助和指点将不胜感激.

Any help and pointers will be appreciated.

谢谢-

推荐答案

由于动态端点，消息处理越来越慢，我看到在创建和使用动态出站端点时线程拥塞.我使用的是ule子3.3.x，在查看了ule子3.4.x代码后，我意识到动态出站终结点创建的处理更为适当.升级到3.4，问题几乎消失了.

Message processing is getting slow because of dynamic endpoint, I see thread congestion when dynamic outbound endpoint is created and used. I was using mule 3.3.x and after looking at mule 3.4.x code i realized that dynamic outbound endpoint creation is handled more appropriately. upgraded to 3.4 and the issue is almost gone.

这篇关于具有Jms连接器的ule子流，动态出站端点中的线程阻塞的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！