Hi Jens,
exactly, approximately 1.6 msg/s.
The processes responsible for reading the queue is single-threaded??
No, I defined a PooledConnectionFactory with concurrentConsumers 10 and 8
max connections, limited with prefetchSize of 10 to avoid OOM.
In addition, I added a custom parallelProcessing on splitter to speed up
defined like this :
<bean id="myThreadPoolExecutor"
class="java.util.concurrent.ThreadPoolExecutor">
<constructor-arg index="0" value="3"/>
<constructor-arg index="1" value="6"/>
<constructor-arg index="2" value="0"/>
<constructor-arg index="3" value="MILLISECONDS"/>
<constructor-arg index="4"><bean
class="java.util.concurrent.LinkedBlockingQueue"/></constructor-arg>
</bean>
More than 3 => Out Of Memory.
I also defined a throttle policy (to avoid overlaod on RS Service) on route
that consumes the queue but it seems that it is never reached.
<route id="ProcessMessageData_Route" errorHandlerRef="DLQErrorHandler" >
<from
uri="activemq:queue:CBIKIT?destination.consumer.prefetchSize=10" />
<throttle timePeriodMillis="1000" asyncDelayed="true">
<constant>5</constant>
<enrich uri="direct:crm-login"
strategyRef="OAuthStrategy" />
<setHeader
headerName="Content-Type"><constant>application/json</constant></setHeader>
<setHeader
headerName="CamelHttpMethod"><constant>POST</constant></setHeader>
<setHeader headerName="CompleteActionPath">
<simple>jetty:{{crm.rs.host.name}}:{{crm.rs.port}}{{crm.rs.context.path}}{{crm.rs.serialnumber.path}}?httpClient.idleTimeout=30000&jettyHttpBindingRef=CustomJettyHttpBinding</simple>
</setHeader>
<log message="Sending SerialNumber from CBIKIT
to CRT -
${in.header.CamelSplitIndex}!" loggingLevel="DEBUG" />
<log message="Sending SerialNumber from CBIKIT
to CRT
${in.header.CamelSplitIndex} - ${body}!" />
<recipientList>
<header>CompleteActionPath</header>
</recipientList>
<log message="Message Received (JSON Format)
${in.header.CamelSplitIndex}: ${body}" />
<unmarshal ref="Gson" />
<choice>
<when>
<simple>${in.header.Result} ==
'ERROR'</simple>
<log message="CRM Response:
${body}" loggingLevel="ERROR"/>
<bean ref="FailureMessageBean"
method="enrichByCRMError" />
<to
uri="direct:messageDiscarded"/>
</when>
</choice>
</throttle>
</route>
Current Active MQ configuration (I'm studying Active MQ guide to improve
performance):
<policyEntry queue="CBIKIT" producerFlowControl="true"
optimizedDispatch="true" reduceMemoryFootprint="true" memoryLimit="256mb" >
</policyEntry>
PersistenceAdapter
<kahaDB directory="${data}/kahadb" indexCacheSize="40000"
indexWriteBatchSize="4000" journalMaxFileLength="512mb"/>
Thanks a lot for your support.
Best Regards
Michele
--
View this message in context:
http://camel.465427.n5.nabble.com/Best-Strategy-to-process-a-large-number-of-rows-in-File-tp5779856p5781173.html
Sent from the Camel - Users mailing list archive at Nabble.com.
