Skip to content

Instantly share code, notes, and snippets.

@dsyer
Created July 12, 2010 10:57
Show Gist options
  • Star 3 You must be signed in to star a gist
  • Fork 2 You must be signed in to fork a gist
  • Save dsyer/472357 to your computer and use it in GitHub Desktop.
Save dsyer/472357 to your computer and use it in GitHub Desktop.
A sample using JDBC polling to test scalability with Spring Integration
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" path="src/main/resources"/>
<classpathentry kind="src" path="src/main/java"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="src" path="src/test/resources"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>
derby-home
derby.log
lib
target
.settings
integration-repo
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>jdbc-test</name>
<comment>This project is a simple template for a jar utility using Spring.
</comment>
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.jdt.core.javabuilder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.springframework.ide.eclipse.core.springbuilder</name>
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.maven.ide.eclipse.maven2Builder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.springframework.ide.eclipse.core.springnature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
</natures>
</projectDescription>
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<version>1</version>
<pluginVersion><![CDATA[2.3.3.201005160334-M1]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[true]]></enableImports>
<configs>
<config>src/main/resources/META-INF/spring/app-context.xml</config>
</configs>
<configSets>
<configSet>
<name><![CDATA[test]]></name>
<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
<incomplete>false</incomplete>
<configs>
</configs>
</configSet>
</configSets>
</beansProjectDescription>
This is a scratch project for demonstrating the use of Spring Integration JDBC adapters to
create independent competing consumers ontop of a relational database.
To use it, clone from gist and build with Maven
+---
$ git clone git://gist.github.com/472357.git jdbc-test
$ cd jdbc-test
$ mvn test
+---
There are configurations for oracle, mysql and derby (use -Denvironment=oracle on the command line
or -DENVIRONMENT=oracle in Eclipse), but only oracle works, presumably because of a
concurrency issue with the others. In the oracle you will see exceptions, but they are just
telling us that a transaction failed because it couldn't lock the table - fortunately most
transactions succeed, so the test passes in the end and throughput isn't bad.
To initialize Oracle you need a user with permission to create tables and sequences. The
default configuration uses a local XE instance with user SPRING:
+---
create user SPRING identified by SPRING;
CREATE ROLE conn;
GRANT CREATE session, CREATE seqence, CREATE trigger,
CREATE table, CREATE view,
CREATE procedure, CREATE synonym,
ALTER any table, ALTER any sequence,
DROP any table, DROP any view, DROP any procedure, DROP any synonym,
DROP any sequence, DROP any trigger
TO conn;
GRANT conn to SPRING;
alter user SPRING default tablespace USERS;
grant unlimited tablespace to SPRING;
+---
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:task="http://www.springframework.org/schema/task" xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd">
<int:channel id="insert" />
<chain input-channel="target" xmlns="http://www.springframework.org/schema/integration">
<splitter expression="payload" />
<transformer expression="payload['name']" />
<service-activator ref="service" method="process" />
</chain>
<poller default="true" max-messages-per-poll="1" receive-timeout="5000"
fixed-delay="100"
xmlns="http://www.springframework.org/schema/integration">
<transactional isolation="SERIALIZABLE" />
</poller>
<int-jdbc:inbound-channel-adapter channel="target" query="${int.poll.query}"
select-sql-parameter-source="parameterSource" data-source="dataSource" max-rows-per-poll="2"
update="update item set status=10 where id in (:id)" />
<int-jdbc:inbound-channel-adapter channel="target" query="${int.poll.query}"
select-sql-parameter-source="parameterSource" data-source="dataSource" max-rows-per-poll="2"
update="update item set status=10 where id in (:id)" />
<int-jdbc:inbound-channel-adapter channel="target" query="${int.poll.query}"
select-sql-parameter-source="parameterSource" data-source="dataSource" max-rows-per-poll="2"
update="update item set status=10 where id in (:id)" />
<int-jdbc:inbound-channel-adapter channel="target" query="${int.poll.query}"
select-sql-parameter-source="parameterSource" data-source="dataSource" max-rows-per-poll="2"
update="update item set status=10 where id in (:id)" />
<int-jdbc:outbound-channel-adapter channel="insert" query="${int.insert.query}"
data-source="dataSource" />
<int:gateway id="gateway" default-request-channel="insert" service-interface="org.springframework.sample.Sender" />
<int-jdbc:inbound-channel-adapter channel="keys" query="${int.keys.query}"
data-source="dataSource">
<poller max-messages-per-poll="1" receive-timeout="5000" xmlns="http://www.springframework.org/schema/integration">
<interval-trigger interval="2000" initial-delay="0"/>
</poller>
</int-jdbc:inbound-channel-adapter>
<int:service-activator ref="parameterSource" input-channel="keys" />
<bean id="parameterSource" class="org.springframework.sample.KeySqlParameterSource" />
<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="${int.jdbc.driver}" />
<property name="url" value="${int.jdbc.url}" />
<property name="username" value="${int.jdbc.user}" />
<property name="password" value="${int.jdbc.password}" />
</bean>
<jdbc:initialize-database data-source="dataSource" ignore-failures="DROPS">
<jdbc:script location="${int.drop.script}" />
<jdbc:script location="${int.schema.script}" />
</jdbc:initialize-database>
<bean id="placeholderProperties" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="location" value="classpath:int-${ENVIRONMENT:derby}.properties" />
<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
<property name="ignoreUnresolvablePlaceholders" value="true" />
<property name="order" value="1" />
</bean>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="service" class="org.springframework.sample.ExampleService" />
<bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
<property name="poolSize" value="${int.scheduler.pool.size}" />
</bean>
</beans>
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
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.xsd">
<import resource="classpath:/META-INF/spring/app-context.xml"/>
</beans>
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed 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.springframework.sample;
import static org.junit.Assert.assertEquals;
import java.util.Random;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StopWatch;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class ExampleConfigurationTests {
private static Log logger = LogFactory.getLog(ExampleConfigurationTests.class);
@Autowired
private ExampleService service;
@Autowired
private Sender gateway;
@Autowired
private PlatformTransactionManager transactionManager;
private SimpleJdbcTemplate jdbcTemplate;
@Autowired
public void setDataSource(DataSource dataSource) {
this.jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
private StopWatch stopWatch = new StopWatch();
@Test
public void testSimpleProperties() throws Exception {
int count = 50;
service.expect(count);
send(count);
service.await(10000);
stopWatch.stop();
logger.info("**** Milliseconds per item: "+stopWatch.getTotalTimeMillis()/count);
assertEquals(count, jdbcTemplate.queryForInt("select count(id) from item where status=10"));
}
private void send(final int count) {
final String[] keys = new String[] {"A", "B", "C", "D", "E"};
final Random random = new Random();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
public Void doInTransaction(TransactionStatus status) {
for (int i = 0; i < count; i++) {
String key = keys[random.nextInt(keys.length)];
gateway.send(new Item(key, "foo" + i));
}
stopWatch.start();
return null;
}
});
}
}
package org.springframework.sample;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* {@link Service} with hard-coded input data.
*/
public class ExampleService implements Service {
private static Log logger = LogFactory.getLog(ExampleService.class);
private CountDownLatch latch = new CountDownLatch(0);
public void expect(int count) {
latch = new CountDownLatch(count);
}
public void await(long timeout) throws InterruptedException, TimeoutException {
if (!latch.await(timeout, TimeUnit.MILLISECONDS)) {
throw new TimeoutException("Timed out waiting for messages");
}
}
public void process(String input) {
logger.debug("Processing: "+input);
try {
Thread.sleep(20);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException();
}
latch.countDown();
}
}
# Placeholders for Derby:
int.jdbc.driver=org.apache.derby.jdbc.EmbeddedDriver
int.jdbc.url=jdbc:derby:derby-home/test;create=true
int.jdbc.user=app
int.jdbc.password=
int.drop.script=classpath:/schema-drop-derby.sql
int.schema.script=classpath:/schema-derby.sql
int.insert.query=insert into item (bus_key,name,status) values (:payload.key,:payload.name, 2)
int.poll.query=select * from item where bus_key=:key and status=2
int.keys.query=select distinct bus_key from item where status=2
int.scheduler.pool.size=1
# Placeholders for MySQL:
int.jdbc.driver=com.mysql.jdbc.Driver
int.jdbc.url=jdbc:mysql://localhost/test
int.jdbc.user=test
int.jdbc.password=test
int.drop.script=classpath:/schema-drop-mysql.sql
int.schema.script=classpath:/schema-mysql.sql
int.insert.query=insert into item (bus_key,name,status) values (:payload.key, :payload.name, 2)
int.poll.query=select id,name,status from item where bus_key=:key and status=2 limit 3
int.keys.query=select distinct bus_key from item where status=2
int.scheduler.pool.size=10
# Placeholders for Oracle:
int.jdbc.driver=oracle.jdbc.OracleDriver
int.jdbc.url=jdbc:oracle:thin:@localhost:1521:xe
int.jdbc.user=spring
int.jdbc.password=spring
int.drop.script=classpath:/schema-drop-oracle.sql
int.schema.script=classpath:/schema-oracle.sql
int.insert.query=insert into item (id,bus_key,name,status) values (item_seq.nextval,:payload.key, :payload.name, 2)
int.poll.query=select * from (select id,name,status,rownum as tmp_row from item where bus_key=:key and status=2) where tmp_row<3
int.keys.query=select distinct bus_key from item where status=2
int.scheduler.pool.size=10
/*
* Copyright 2002-2008 the original author or authors.
*
* Licensed 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.springframework.sample;
/**
* @author Dave Syer
*
*/
public class Item {
private final String name;
private final String key;
public Item(String key, String name) {
this.key = key;
this.name = name;
}
/**
* @return the key
*/
public String getKey() {
return key;
}
/**
* @return the name
*/
public String getName() {
return name;
}
}
package org.springframework.sample;
import java.util.List;
import java.util.Map;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.jdbc.core.namedparam.AbstractSqlParameterSource;
public class KeySqlParameterSource extends AbstractSqlParameterSource {
private static Log logger = LogFactory.getLog(KeySqlParameterSource.class);
// Seed the keys with sensible values to start with
private volatile String[] keys = new String[] {"A", "B", "C", "D", "E"};
final Random random = new Random();
@ServiceActivator
public void setKeys(List<Map<String, String>> keys) {
logger.debug("Set keys:" + keys);
String[] result = new String[keys.size()];
int count = 0;
for (Map<String, String> key : keys) {
result[count++] = key.get("BUS_KEY");
}
this.keys = result;
}
public Object getValue(String paramName) throws IllegalArgumentException {
String[] keys = this.keys;
return keys[random.nextInt(keys.length)];
}
public boolean hasValue(String paramName) {
return "key".equals(paramName);
}
}
log4j.rootCategory=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
#log4j.category.org.springframework.jdbc=DEBUG
log4j.category.org.springframework.jdbc.datasource.init=ERROR
log4j.category.org.springframework.transaction=INFO
#log4j.category.org.springframework.integration=DEBUG
log4j.category.org.springframework.scheduling.support=FATAL
log4j.category.org.springframework.sample=DEBUG
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.samples.spring</groupId>
<artifactId>jdbc-test</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>jdbc-test</name>
<url>http://www.springframework.org</url>
<description>
<![CDATA[
This project is a minimal jar utility with Spring configuration.
]]>
</description>
<properties>
<environment>oracle</environment>
<maven.test.failure.ignore>true</maven.test.failure.ignore>
<spring.framework.version>3.0.5.RELEASE</spring.framework.version>
<spring.integration.version>2.0.1.RELEASE</spring.integration.version>
</properties>
<profiles>
<profile>
<id>strict</id>
<properties>
<maven.test.failure.ignore>false</maven.test.failure.ignore>
</properties>
</profile>
</profiles>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.8.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring.framework.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
<version>${spring.framework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
<version>${spring.framework.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>${spring.integration.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jdbc</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjrt</artifactId>
<version>1.6.8</version>
</dependency>
<dependency>
<groupId>org.aspectj</groupId>
<artifactId>aspectjweaver</artifactId>
<version>1.6.8</version>
</dependency>
<dependency>
<groupId>org.apache.derby</groupId>
<artifactId>derby</artifactId>
<version>10.6.1.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.6</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>com.oracle.jdbc</groupId>
<artifactId>com.springsource.oracle.jdbc</artifactId>
<version>10.2.0.2</version>
<optional>true</optional>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>commons-dbcp</groupId>
<artifactId>commons-dbcp</artifactId>
<version>1.2.2</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
</dependency>
</dependencies>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>com.springsource.bundlor</groupId>
<artifactId>com.springsource.bundlor.maven</artifactId>
<version>1.0.0.M1B</version>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<inherited>false</inherited>
<configuration>
<descriptorRefs>
<descriptorRef>project</descriptorRef>
</descriptorRefs>
</configuration>
</plugin>
</plugins>
</pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.5</source>
<target>1.5</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<!--forkMode>pertest</forkMode -->
<includes>
<include>**/*Tests.java</include>
</includes>
<excludes>
<exclude>**/Abstract*.java</exclude>
</excludes>
<systemProperties>
<property>
<name>ENVIRONMENT</name>
<value>${environment}</value>
</property>
</systemProperties>
</configuration>
</plugin>
</plugins>
</build>
<pluginRepositories>
<pluginRepository>
<id>Codehaus</id>
<url>http://repository.codehaus.org/</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
<repositories>
<repository>
<id>repository.springframework.ebr.external</id>
<name>Spring Framework EBR External Repository</name>
<url>http://repository.springsource.com/maven/bundles/external/</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<distributionManagement>
<downloadUrl>http://www.springframework.org/download</downloadUrl>
<site>
<id>staging</id>
<url>file:///${user.dir}/target/staging/org.springframework.batch.archetype/${pom.artifactId}</url>
</site>
<repository>
<id>spring-release</id>
<name>Spring Release Repository</name>
<url>file:///${user.dir}/target/staging/release</url>
</repository>
<snapshotRepository>
<id>spring-snapshot</id>
<name>Spring Snapshot Repository</name>
<url>file:///${user.dir}/target/staging/snapshot</url>
</snapshotRepository>
</distributionManagement>
</project>
create table item(id int PRIMARY KEY GENERATED ALWAYS AS IDENTITY, bus_key char(1), name varchar(20), status int);
create index item_index on item(status,bus_key);
drop table item;
drop sequence item_seq;
drop table item;
drop sequence item_seq;
create table item(id BIGINT PRIMARY KEY AUTO_INCREMENT, bus_key char(1), name varchar(20), status BIGINT);
create index item_index on item(status,bus_key);
create table item(id NUMBER(19) PRIMARY KEY, bus_key char(1), name varchar(20), status NUMBER(19)) INITRANS 100;
create sequence item_seq;
create index item_index on item(status,bus_key) INITRANS 100;
package org.springframework.sample;
public interface Sender {
void send(Item input);
}
package org.springframework.sample;
public interface Service {
void process(String input);
}
<?xml version="1.0" encoding="ISO-8859-1"?>
<project name="Spring Batch: ${project.name}">
<bannerLeft>
<name>Spring Batch: ${project.name}</name>
<href>index.html</href>
</bannerLeft>
<skin>
<groupId>org.springframework.maven.skins</groupId>
<artifactId>maven-spring-skin</artifactId>
<version>1.0.5</version>
</skin>
<body>
<links>
<item name="${project.name}" href="index.html"/>
</links>
<menu ref="reports"/>
</body>
</project>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment