activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From James Strachan <james.strac...@gmail.com>
Subject Re: ActiveMQ to Websphere MQ
Date Sat, 14 Aug 2010 07:45:36 GMT
If you want to concurrently consume from ActiveMQ (or any JMS) with
Camel, just set the concurrentConsumers property on the URI (or
configure the component/endpoint).

See
http://camel.apache.org/jms.html

Note that concurrent consumers break order of messages on a queue. To
fix this using ActiveMQ you can use Message Groups
http://activemq.apache.org/message-groups.html

On 31 March 2010 16:03, sweenma <mark.sweeney@verizonwireless.com> wrote:
>
> Hi,
> I am new to ActiveMQ and Q'ing in general, so I will do my best to explain
> the issue and ask for your patience. I followed the suggestions of many
> postings on how to get ActiveMQ to forward msg's to Websphere MQ using
> camel. I got it to work, however the amount of time it takes to forward the
> msg from ActiveMQ to Websphere MQ is very high, around 2 secs. I enabled the
> logging and I noticed that it appears that the communication between
> ActiveMQ and WebSphere MQ runs 1 thread at a time. For example, I have a
> simple multithreaded producer that puts msg's in to the ActiveMQ. ActiveMQ
> is configured to forward the msg's to WebSphere MQ and then I have a simple
> MDB listener to read the msg's from the WebSphere MQ. With logging enabled I
> can see a msg being processed by a thread and then another msg being
> processed by another thread after the first thread completes. Here is a
> samle of the log:
>
> 2010-03-31 08:32:44,743 [DefaultMessageListenerContainer-29]
> org.apache.activemq.ActiveMQMessageConsumer - ID:WNJ10002LDVSBAG
> -1559-1270038729745-2:0:1:1 received message: MessageDispatch {commandId =
> 0, responseRequired = false, consumerId =
> ID:WNJ10002LDVSBAG-1559-1270038729745-2:0:1:1,
>
> lots of other log entries with the same thread id
> DefaultMessageListenerContainer-29 followed by the next msg log entry:
> 2010-03-31 08:32:46,539 [DefaultMessageListenerContainer-30]
> org.apache.activemq.ActiveMQMessageConsumer - ID:WNJ10002LDVSBAG
> -1559-1270038729745-2:0:1:1 received message: MessageDispatch {commandId =
> 0, responseRequired = false, consumerId =
> ID:WNJ10002LDVSBAG-1559-1270038729745-2:0:1:1
>
> and so on for each msg that was put into the ActiveMQ queue. I noticed that
> the ID for each ActiveMQMessageConsumer is the same, however I am not sure
> if this is relavant or not.
> I am using activemq V5.2.0I with Java V1.5. I pasted a copy of the
> activemq.xml file below. I am really stuck here and would appreciate any
> help. TIA.
>
> <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
>  http://activemq.apache.org/camel/schema/spring
> http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
>
>    <!-- Allows us to use system properties as variables in this
> configuration file -->
>    <bean
> class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
>         <property name="locations">
>
> <value>file:///${activemq.base}/conf/credentials.properties</value>
>         </property>
>    </bean>
>
>    <broker xmlns="http://activemq.apache.org/schema/core"
> brokerName="mylaptop" persistent="false"
> dataDirectory="${activemq.base}/data" useJmx="false">
>
>        <!-- Destination specific policies using destination names or
> wildcards -->
>        <destinationPolicy>
>            <policyMap>
>                <policyEntries>
>                    <policyEntry queue=">" memoryLimit="5mb"/>
>                    <policyEntry topic=">" memoryLimit="5mb">
>                      <!-- you can add other policies too such as these
>                        <dispatchPolicy>
>                            <strictOrderDispatchPolicy/>
>                        </dispatchPolicy>
>                        <subscriptionRecoveryPolicy>
>                            <lastImageSubscriptionRecoveryPolicy/>
>                        </subscriptionRecoveryPolicy>
>                      -->
>                    </policyEntry>
>                </policyEntries>
>            </policyMap>
>        </destinationPolicy>
>
>        <destinations>
>            <queue name="testRIMQ" physicalName="RIMQ" />
>        </destinations>
>
>        <!-- Use the following to configure how ActiveMQ is exposed in JMX
> -->
>        <managementContext>
>            <managementContext createConnector="false"/>
>        </managementContext>
>
>        <!-- The store and forward broker networks ActiveMQ will listen to
> -->
>        <networkConnectors>
>        </networkConnectors>
>
>        <persistenceAdapter>
>            <amqPersistenceAdapter syncOnWrite="false"
> directory="${activemq.base}/data" indexPageSize="16kb"
> maxFileLength="32mb"/>
>        </persistenceAdapter>
>
>
>        <sslContext>
>            <sslContext keyStore="file:${activemq.base}/conf/broker.ks"
> keyStorePassword="password"
> trustStore="file:${activemq.base}/conf/broker.ts"
> trustStorePassword="password"/>
>        </sslContext>
>
>        <!--  The maximum about of space the broker will use before slowing
> down producers -->
>        <systemUsage>
>            <systemUsage>
>                <memoryUsage>
>                    <memoryUsage limit="20 mb"/>
>                </memoryUsage>
>                <storeUsage>
>                    <storeUsage limit="1 gb" name="foo"/>
>                </storeUsage>
>                <tempUsage>
>                    <tempUsage limit="100 mb"/>
>                </tempUsage>
>            </systemUsage>
>        </systemUsage>
>
>
>        <!-- The transport connectors ActiveMQ will listen to -->
>        <transportConnectors>
>            <transportConnector name="openwire" uri="tcp://localhost:61616"
> />
>        </transportConnectors>
>
>    </broker>
>
>    <camelContext id="camel"
> xmlns="http://activemq.apache.org/camel/schema/spring">
>        <route>
>            <from uri="activemq:RIMQ"/>
>            <to uri="webspheremq:queue:DVST.TEST.QUEUE"/>
>        </route>
>    </camelContext>
>
>
>
>    <bean id="remoteBroker"
> class="org.apache.activemq.camel.component.ActiveMQComponent" >
>        <property name="connectionFactory">
>          <bean class="org.apache.activemq.ActiveMQConnectionFactory">
>            <property name="brokerURL"
> value="failover:(tcp://host:port)?maxReconnectAttempts=0" />
>          </bean>
>        </property>
>    </bean>
>
>    <bean id="webspheremq"
> class="org.apache.camel.component.jms.JmsComponent">
>        <property name="configuration">
>            <bean class="org.apache.camel.component.jms.JmsConfiguration">
>                <property name="connectionFactory">
>                    <bean class="com.ibm.mq.jms.MQQueueConnectionFactory">
>                        <property name="hostName" value="...."/>
>                        <property name="port" value="1...."/>
>                        <property name="queueManager" value="Q1T2"/>
>                        <property name="channel"
> value="DVST.COMMON.SVRTST"/>
>                        <property name="transportType" value="1"/>
>                    </bean>
>                </property>
>            </bean>
>        </property>
>    </bean>
>
>    <!-- An embedded servlet engine for serving up the Admin console -->
>    <jetty xmlns="http://mortbay.com/schemas/jetty/1.0">
>        <connectors>
>            <nioConnector port="8161"/>
>        </connectors>
>
>        <handlers>
>            <webAppContext contextPath="/admin"
> resourceBase="${activemq.base}/webapps/admin" logUrlOnStart="true"/>
>            <webAppContext contextPath="/demo"
> resourceBase="${activemq.base}/webapps/demo" logUrlOnStart="true"/>
>            <webAppContext contextPath="/fileserver"
> resourceBase="${activemq.base}/webapps/fileserver" logUrlOnStart="true"/>
>        </handlers>
>    </jetty>
>
>    <!--  This xbean configuration file supports all the standard spring xml
> configuration options -->
>
>
> </beans>
>
> --
> View this message in context: http://old.nabble.com/ActiveMQ-to-Websphere-MQ-tp28097168p28097168.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
>



-- 
James
-------
http://macstrac.blogspot.com/

Open Source Integration
http://fusesource.com/

Mime
View raw message