mirror of https://github.com/apache/activemq.git
test to explore concurrent transacted consumption with camel and prefetch implications of pooling
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1162061 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully
parent
d8d20f5621
commit
86dc4d5c34
|
|
@ -0,0 +1,123 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.apache.activemq.camel;
|
||||
|
||||
import javax.jms.Connection;
|
||||
import javax.jms.MessageProducer;
|
||||
import javax.jms.Session;
|
||||
import javax.jms.TextMessage;
|
||||
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||
import org.apache.activemq.broker.BrokerService;
|
||||
import org.apache.activemq.broker.region.policy.PolicyEntry;
|
||||
import org.apache.activemq.broker.region.policy.PolicyMap;
|
||||
import org.apache.activemq.command.ActiveMQQueue;
|
||||
import org.apache.activemq.command.ActiveMQTextMessage;
|
||||
import org.apache.activemq.util.Wait;
|
||||
import org.apache.camel.Exchange;
|
||||
import org.apache.camel.Processor;
|
||||
import org.apache.camel.component.jms.JmsMessage;
|
||||
import org.apache.camel.test.junit4.CamelSpringTestSupport;
|
||||
import org.junit.Test;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.context.support.AbstractXmlApplicationContext;
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
|
||||
public class TransactedConsumeTest extends CamelSpringTestSupport {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TransactedConsumeTest.class);
|
||||
BrokerService broker = null;
|
||||
int messageCount = 1000;
|
||||
|
||||
@Test
|
||||
public void testConsume() throws Exception {
|
||||
|
||||
LOG.info("Wait for dequeue message...");
|
||||
|
||||
assertTrue(Wait.waitFor(new Wait.Condition() {
|
||||
@Override
|
||||
public boolean isSatisified() throws Exception {
|
||||
return broker.getAdminView().getTotalDequeueCount() >= messageCount;
|
||||
}
|
||||
}, 20 * 60 * 1000));
|
||||
}
|
||||
|
||||
private void sendJMSMessageToKickOffRoute() throws Exception {
|
||||
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://test");
|
||||
factory.setWatchTopicAdvisories(false);
|
||||
Connection connection = factory.createConnection();
|
||||
connection.start();
|
||||
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||
MessageProducer producer = session.createProducer(new ActiveMQQueue("scp_transacted"));
|
||||
for (int i=0; i<messageCount;i++) {
|
||||
TextMessage message = session.createTextMessage("Some Text, messageCount:" + i);
|
||||
message.setJMSCorrelationID("pleaseCorrelate");
|
||||
producer.send(message);
|
||||
}
|
||||
LOG.info("Sent: " + messageCount);
|
||||
connection.close();
|
||||
}
|
||||
|
||||
private BrokerService createBroker(boolean deleteAllMessages) throws Exception {
|
||||
BrokerService brokerService = new BrokerService();
|
||||
brokerService.setDeleteAllMessagesOnStartup(deleteAllMessages);
|
||||
brokerService.setBrokerName("test");
|
||||
|
||||
PolicyMap policyMap = new PolicyMap();
|
||||
PolicyEntry defaultPolicy = new PolicyEntry();
|
||||
// defaultPolicy.setStrictOrderDispatch(false);
|
||||
policyMap.setDefaultEntry(defaultPolicy);
|
||||
brokerService.setDestinationPolicy(policyMap);
|
||||
|
||||
brokerService.setAdvisorySupport(false);
|
||||
brokerService.setDataDirectory("target/data");
|
||||
brokerService.addConnector("tcp://localhost:61616");
|
||||
return brokerService;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractXmlApplicationContext createApplicationContext() {
|
||||
|
||||
deleteDirectory("target/data");
|
||||
|
||||
// make broker available to recovery processing on app context start
|
||||
try {
|
||||
broker = createBroker(true);
|
||||
broker.start();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException("Failed to start broker", e);
|
||||
}
|
||||
|
||||
try {
|
||||
sendJMSMessageToKickOffRoute();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException("Failed to fill q", e);
|
||||
}
|
||||
|
||||
return new ClassPathXmlApplicationContext("org/apache/activemq/camel/transactedconsume.xml");
|
||||
}
|
||||
|
||||
static class ConnectionLog implements Processor {
|
||||
|
||||
@Override
|
||||
public void process(Exchange exchange) throws Exception {
|
||||
ActiveMQTextMessage m = (ActiveMQTextMessage) ((JmsMessage)exchange.getIn()).getJmsMessage();
|
||||
Thread.currentThread().sleep(10);
|
||||
LOG.info("received on " + m.getConnection().toString());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
<!--
|
||||
Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
contributor license agreements. See the NOTICE file distributed with
|
||||
this work for additional information regarding copyright ownership.
|
||||
The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
(the "License"); you may not use this file except in compliance with
|
||||
the License. You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
-->
|
||||
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:context="http://www.springframework.org/schema/context"
|
||||
xsi:schemaLocation="
|
||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
|
||||
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
|
||||
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
|
||||
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
|
||||
">
|
||||
|
||||
<context:annotation-config/>
|
||||
|
||||
<bean id="vhfBatchListenerJMSConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
|
||||
<property name="brokerURL" value="tcp://localhost:61616?jms.prefetchPolicy.all=1"/>
|
||||
</bean>
|
||||
|
||||
<bean id="vhfBatchListenerPooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
|
||||
<property name="maxConnections" value="10"/>
|
||||
<property name="maximumActive" value="10"/>
|
||||
<property name="connectionFactory" ref="vhfBatchListenerJMSConnectionFactory"/>
|
||||
</bean>
|
||||
|
||||
<!-- bean id="vhfBatchListenerSingleConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
|
||||
<property name="reconnectOnException" value="true" />
|
||||
<property name="targetConnectionFactory" ref="vhfBatchListenerJMSConnectionFactory" />
|
||||
</bean -->
|
||||
|
||||
<!-- JMS Transaction manager -->
|
||||
<bean id="vhfBatchListenerJMSTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
|
||||
<property name="connectionFactory" ref="vhfBatchListenerPooledConnectionFactory"/>
|
||||
</bean>
|
||||
|
||||
<!-- JMS Configuration -->
|
||||
<bean id="vhfBatchListenerJMSConfig" class="org.apache.camel.component.jms.JmsConfiguration">
|
||||
<property name="connectionFactory" ref="vhfBatchListenerPooledConnectionFactory"/>
|
||||
<property name="transactionManager" ref="vhfBatchListenerJMSTransactionManager"/>
|
||||
<property name="transacted" value="true"/>
|
||||
<property name="concurrentConsumers" value="10"/>
|
||||
<property name="cacheLevelName" value="CACHE_CONSUMER"/>
|
||||
</bean>
|
||||
|
||||
<!-- JMS Transaction policy -->
|
||||
<bean id="vhfBatchListenerTransaction" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
|
||||
<property name="transactionManager" ref="vhfBatchListenerJMSTransactionManager"/>
|
||||
</bean>
|
||||
|
||||
<!-- ActiveMQ component -->
|
||||
<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
|
||||
<property name="configuration" ref="vhfBatchListenerJMSConfig"/>
|
||||
</bean>
|
||||
|
||||
<camelContext xmlns="http://camel.apache.org/schema/spring">
|
||||
<route>
|
||||
<from uri="activemq:queue:scp_transacted"/>
|
||||
<!-- transacted /-->
|
||||
<process ref="connectonLog"/>
|
||||
</route>
|
||||
</camelContext>
|
||||
|
||||
<bean id="connectonLog" class="org.apache.activemq.camel.TransactedConsumeTest.ConnectionLog"/>
|
||||
</beans>
|
||||
Loading…
Reference in New Issue