diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java new file mode 100644 index 000000000..7bd6b2fec --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java @@ -0,0 +1,70 @@ +/* + * Copyright 2015 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.batch.item.support; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemStreamReader; +import org.springframework.batch.item.NonTransientResourceException; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; + +/** + * + * This is a simple ItemStreamReader decorator with a synchronized ItemReader.read() + * method - which makes a non-thread-safe ItemReader thread-safe. + * + * However, if reprocessing an item is problematic then using this will make a job not + * restartable. If a restartable job is desired in that case, then further co-ordination + * between the read and close methods needs to be implemented. + * + * Here are some links about the motivation behind this class: + * - http://projects.spring.io/spring-batch/faq.html#threading-reader} + * - http://stackoverflow.com/a/20002493/2910265} + * + * @author Matthew Ouyang + * @since 3.0 + * + * @param + */ +public class SynchronizedItemStreamReader implements ItemStream, ItemReader { + + ItemStreamReader itemStreamReader; + + public void setItemStreamReader(ItemStreamReader itemStreamReader) { + this.itemStreamReader = itemStreamReader; + } + + /** + * This delegates to the read method of the itemStreamReader + */ + public synchronized T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException { + return this.itemStreamReader.read(); + } + + public void close() { + this.itemStreamReader.close(); + } + + public void open(ExecutionContext executionContext) { + this.itemStreamReader.open(executionContext); + } + + public void update(ExecutionContext executionContext) { + this.itemStreamReader.update(executionContext); + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java new file mode 100644 index 000000000..b10ba64f7 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java @@ -0,0 +1,140 @@ +/* + * Copyright 2015 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.batch.item.support; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.HashSet; +import java.util.Set; + +import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamReader; +import org.springframework.batch.item.NonTransientResourceException; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; + +/** + * + * @author Matthew Ouyang + * + */ +public class SynchronizedItemStreamReaderTests { + + /** + * A simple class used to test the SynchronizedItemStreamReader. It simply returns + * the number of times the read method has been called, manages some state variables + * and updates an ExecutionContext. + * + * @author Matthew Ouyang + * + */ + private class TestItemReader extends AbstractItemStreamItemReader implements ItemStreamReader { + + private int cursor = 0; + private boolean isClosed = false; + + public static final String HAS_BEEN_OPENED = "hasBeenOpened"; + public static final String UPDATE_COUNT_KEY = "updateCount"; + + public Integer read() throws Exception, UnexpectedInputException, ParseException, + NonTransientResourceException { + cursor = cursor + 1; + return cursor; + } + + public void close() { + this.isClosed = true; + } + + public void open(ExecutionContext executionContext) { + this.isClosed = false; + executionContext.put(HAS_BEEN_OPENED, true); + executionContext.remove(UPDATE_COUNT_KEY); + } + + public void update(ExecutionContext executionContext) { + + if (!executionContext.containsKey(UPDATE_COUNT_KEY)) { + executionContext.putInt(UPDATE_COUNT_KEY, 0); + } + + executionContext.putInt(UPDATE_COUNT_KEY + , executionContext.getInt(UPDATE_COUNT_KEY) + 1 + ); + } + + public boolean isClosed() { + return this.isClosed; + } + } + + @Test + public void testMultipleThreads() throws Exception { + + // Initialized an ExecutionContext and a SynchronizedItemStreamReader to test. + final ExecutionContext executionContext = new ExecutionContext(); + + final TestItemReader testItemReader = new TestItemReader(); + final SynchronizedItemStreamReader synchronizedItemStreamReader = new SynchronizedItemStreamReader(); + synchronizedItemStreamReader.setItemStreamReader(testItemReader); + + // Open the ItemReader and make sure it's initialized properly. + synchronizedItemStreamReader.open(executionContext); + assertEquals(true, executionContext.get(TestItemReader.HAS_BEEN_OPENED)); + assertFalse(testItemReader.isClosed()); + + /* Set up SIZE threads that read from the reader and updates the execution + * context. + */ + final Set ecSet = new HashSet(); + final int SIZE = 20; + Thread[] threads = new Thread[SIZE]; + for (int i = 0; i < SIZE; i++) { + threads[i] = new Thread() { + public void run() { + try { + ecSet.add(synchronizedItemStreamReader.read()); + synchronizedItemStreamReader.update(executionContext); + } catch (Exception e) { + } finally { + } + } + }; + } + + // Start the threads and block until all threads are done. + for (Thread thread : threads) { + thread.run(); + } + for (Thread thread : threads) { + thread.join(); + } + testItemReader.close(); + + /* Ensure cleanup happens as expected: status variable is set correctly and + * ExecutionContext variable is set properly. Lastly, the Set should + * have 1 to 20 which may not always be the case if the read is not synchronized. + */ + for (int i = 1; i <= SIZE; i++) { + assertTrue(ecSet.contains(i)); + } + assertTrue(testItemReader.isClosed()); + assertEquals(SIZE, executionContext.getInt(TestItemReader.UPDATE_COUNT_KEY)); + } +}