Increasing speed of message forwarding

[pollotek]

I have these app servers running Tomcat enqueueing messages to an embedded
broker. These messages get forwarded to a stand alone broker. There is
another machine connected to the stand alone broker consuming messages. The
embedded broker and the stand alone broker are persistent. 

I have this queue that seems to grow faster in the embedded broker than
messages getting forwarding to the stand alone broker. I can clearly see
dispatch count is smaller than enqueue count. I see there is only one
consumer connected for this queue on the embedded broker going to the stand
alone broker.

I see the stand alone broker is about 95% idle, meaning it's not at max
capacity and iowait 0.4%. Is there a configuration parameter to make the
stand alone broker increase the number of threads/consumers reading from the
embedded brokers? I fear that even if I add more consumer threads in my
consumer servers the rate at which messages are being forwarded to the stand
alone broker will not be affected. 

I'm running ActiveMQ 5.6.0 on all servers.

Any ideas?

My stand alone broker config is:

<beans
  xmlns="http://www.springframework.org/schema/beans";
  xmlns:amq="http://activemq.apache.org/schema/core";
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
  xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd";>

    <bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>

    <broker xmlns="http://activemq.apache.org/schema/core";
            brokerName="broker"
            persistent="true"
            dataDirectory="/var/gluster/activemq/data"
            useJmx="true">

        
        <destinationPolicy>
            <policyMap>
                <policyEntries>
                    
                    <policyEntry queue=">">
                        <deadLetterStrategy>
                            
                            <individualDeadLetterStrategy queuePrefix="DLQ."
useQueueForQueueMessages="true" />
                        </deadLetterStrategy>
                    </policyEntry>
                </policyEntries>
            </policyMap>
        </destinationPolicy>

        <managementContext>
            <managementContext createConnector="true" connectorPort="1199"/>
        </managementContext>

        
        <persistenceAdapter>
            <kahaDB directory="/var/gluster/activemq/data/kahadb"/>
        </persistenceAdapter>

        
        <transportConnectors>
            <transportConnector name="openwire"
uri="tcp://mqbroker01:61616"/>
        </transportConnectors>

    </broker>



My embedded broker configuration:


<beans xmlns="http://www.springframework.org/schema/beans";
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
       xmlns:p="http://www.springframework.org/schema/p";
       xmlns:context="http://www.springframework.org/schema/context";
       xmlns:jms="http://www.springframework.org/schema/jms";
       xmlns:amq="http://activemq.apache.org/schema/core";
       xsi:schemaLocation="http://www.springframework.org/schema/beans
                          
http://www.springframework.org/schema/beans/spring-beans.xsd
                           http://www.springframework.org/schema/context
                          
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
                           http://www.springframework.org/schema/jms
                          
http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
                           http://activemq.apache.org/schema/core
                          
http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd";>


    
    <bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"
>
        <property name="placeholderPrefix" value="_STARTPREFIX"/>
                <property name="placeholderSuffix" value="_ENDSUFFIX"/>
    </bean>

    
    <amq:broker
dataDirectory="_STARTPREFIXactivemq.directory_ENDSUFFIX/data"
id="broker-_STARTPREFIXactivemq.brokerid_ENDSUFFIX" persistent="true">

        
        <amq:plugins>
            <amq:statisticsBrokerPlugin/>
        </amq:plugins>


        <amq:destinationPolicy>
            <amq:policyMap>
                <amq:policyEntries>
                    
                    <amq:policyEntry queue=">">
                        <amq:deadLetterStrategy>
                            
                            <amq:individualDeadLetterStrategy
queuePrefix="DLQ." useQueueForQueueMessages="true" />
                        </amq:deadLetterStrategy>
                    </amq:policyEntry>
                </amq:policyEntries>
            </amq:policyMap>
        </amq:destinationPolicy>


       <amq:persistenceAdapter>
            <amq:kahaDB
directory="_STARTPREFIXactivemq.directory_ENDSUFFIX/data/app/kahadb" />
        </amq:persistenceAdapter>

        <amq:transportConnectors>
            
            <amq:transportConnector uri="vm://localhost?async=true" />
        </amq:transportConnectors>

        
        <amq:networkConnectors>
                       <amq:networkConnector
name="connection_to_nob-_STARTPREFIXactivemq.brokerid_ENDSUFFIX"
                                             
uri="static:failover:(tcp://mqbroker01:61616,tcp://mqbroker02:61616)?randomize=false&amp;maxReconnectAttempts=0"
                                              duplex="false"/>
        </amq:networkConnectors>


    </amq:broker>
</beans>



--
View this message in context: 
http://activemq.2283324.n4.nabble.com/Increasing-speed-of-message-forwarding-tp4656545.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.