From 8423609b43d7770fb12bb4ce64d611cff902fbc3 Mon Sep 17 00:00:00 2001 From: Taeik Lim Date: Sun, 2 Feb 2025 21:16:09 +0900 Subject: [PATCH] Update doc for reader-processor Signed-off-by: Taeik Lim --- ...tem-stream-flux-reader-processor-writer.md | 154 ++++++++++++++++- ...stream-iterable-reader-processor-writer.md | 160 +++++++++++++++++- ...stream-iterator-reader-processor-writer.md | 159 ++++++++++++++++- ...m-stream-simple-reader-processor-writer.md | 145 +++++++++++++++- ...tem-stream-flux-reader-processor-writer.md | 154 ++++++++++++++++- ...stream-iterable-reader-processor-writer.md | 160 +++++++++++++++++- ...stream-iterator-reader-processor-writer.md | 158 ++++++++++++++++- ...m-stream-simple-reader-processor-writer.md | 145 +++++++++++++++- 8 files changed, 1219 insertions(+), 16 deletions(-) diff --git a/doc/en/step/item-stream-flux-reader-processor-writer.md b/doc/en/step/item-stream-flux-reader-processor-writer.md index d665416c..e8fb2d53 100644 --- a/doc/en/step/item-stream-flux-reader-processor-writer.md +++ b/doc/en/step/item-stream-flux-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Create a tasklet without a processor](#create-a-tasklet-without-a-processor) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Use a callback](#use-a-callback) +- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Use a callback](#use-a-callback) + - [Java](#java-3) + - [Kotlin](#kotlin-3) Spring uses a reactive library called [Reactor](https://projectreactor.io/), which provides streams for various types of data using `Flux`. For `ItemReader` in Spring Batch to use the data read using `Flux`, you need to extract each item from `Flux` and return it. @@ -326,9 +329,156 @@ open class TestJobConfig( } ``` +## Create a tasklet without a writer + +If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamFluxReaderProcessor` to define `ItemStreamReader` and `ItemProcessor` in a single class. + +### Java + +In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamFluxReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Flux readFlux(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return Flux.generate(sink -> { + if (count < totalCount) { + sink.next(count); + ++count; + } else { + sink.complete(); + } + }); + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +You can statically import the method of `AdapterFactory` for better readability. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamFluxReaderProcessor { + private var count = 0 + + override fun readFlux(executionContext: ExecutionContext): Flux { + println("totalCount: $totalCount") + return Flux.generate { sink -> + if (count < totalCount) { + sink.next(count) + ++count + } else { + sink.complete() + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Use a callback -You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamFluxReaderProcessorWriter` and `ItemStreamFluxReaderWriter`. You can selectively define a callback method. +You can define a callback method for `ItemStream`. You can selectively define a callback method. ### Java diff --git a/doc/en/step/item-stream-iterable-reader-processor-writer.md b/doc/en/step/item-stream-iterable-reader-processor-writer.md index bd40ebb8..62840c66 100644 --- a/doc/en/step/item-stream-iterable-reader-processor-writer.md +++ b/doc/en/step/item-stream-iterable-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Create a tasklet without a processor](#create-a-tasklet-without-a-processor) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Use a callback](#use-a-callback) +- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Use a callback](#use-a-callback) + - [Java](#java-3) + - [Kotlin](#kotlin-3) A chunk-oriented step in Spring Batch consists of `ItemReader`, `ItemProcessor`, and `ItemWriter`, which are usually defined separately and then assembled to define a step. However, there are some issues with this approach: it is difficult to share data between `ItemReader`, `ItemProcessor`, and `ItemWriter`, and you need to see each respective file to understand the batch flow. Also, if the classes are not reused, they can make the elements of a job less coherent. @@ -328,9 +331,162 @@ open class TestJobConfig( } ``` +## Create a tasklet without a writer + +If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamIterableReaderProcessor` to define `ItemStreamReader` and `ItemProcessor` in a single class. + +### Java + +In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamIterableReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Iterable readIterable(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return () -> new Iterator<>() { + @Override + public boolean hasNext() { + return count < totalCount; + } + + @Override + public Integer next() { + return count++; + } + }; + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +You can statically import the method of `AdapterFactory` for better readability. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamIterableReaderProcessor { + private var count = 0 + + override fun readIterable(executionContext: ExecutionContext): Iterable { + println("totalCount: $totalCount") + return Iterable { + object : Iterator { + override fun hasNext(): Boolean { + return count < totalCount + } + + override fun next(): Int { + return count++ + } + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Use a callback -You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamIterableReaderProcessorWriter` and `ItemStreamIterableReaderWriter`. You can selectively define a callback method. +You can define a callback method for `ItemStream`. You can selectively define a callback method. ### Java diff --git a/doc/en/step/item-stream-iterator-reader-processor-writer.md b/doc/en/step/item-stream-iterator-reader-processor-writer.md index 7d02503e..f6de2d6e 100644 --- a/doc/en/step/item-stream-iterator-reader-processor-writer.md +++ b/doc/en/step/item-stream-iterator-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Create a tasklet without a processor](#create-a-tasklet-without-a-processor) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Use a callback](#use-a-callback) +- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Use a callback](#use-a-callback) + - [Java](#java-3) + - [Kotlin](#kotlin-3) A chunk-oriented step in Spring Batch consists of `ItemReader`, `ItemProcessor`, and `ItemWriter`, which are usually defined separately and then assembled to define a step. However, there are some issues with this approach: it is difficult to share data between `ItemReader`, `ItemProcessor`, and `ItemWriter`, and you need to see each respective file to understand the batch flow. Also, if the classes are not reused, they can make the elements of a job less coherent. @@ -324,9 +327,161 @@ open class TestJobConfig( } ``` +## Create a tasklet without a writer + +If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamIteratorReaderProcessor` to define `ItemStreamReader` and `ItemProcessor` in a single class. + +### Java + +In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamIteratorReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Iterator readIterator(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return new Iterator<>() { + @Override + public boolean hasNext() { + return count < totalCount; + } + + @Override + public Integer next() { + return count++; + } + }; + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} + +``` + +You can statically import the method of `AdapterFactory` for better readability. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamIteratorReaderProcessor { + private var count = 0 + + override fun readIterator(executionContext: ExecutionContext): Iterator { + println("totalCount: $totalCount") + return object : Iterator { + override fun hasNext(): Boolean { + return count < totalCount + } + + override fun next(): Int { + return count++ + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Use a callback -You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamIteratorReaderProcessorWriter` and `ItemStreamIteratorReaderWriter`. You can selectively define a callback method. +You can define a callback method for `ItemStream`. You can selectively define a callback method. ### Java diff --git a/doc/en/step/item-stream-simple-reader-processor-writer.md b/doc/en/step/item-stream-simple-reader-processor-writer.md index ea3d52f0..88b314d2 100644 --- a/doc/en/step/item-stream-simple-reader-processor-writer.md +++ b/doc/en/step/item-stream-simple-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Create a tasklet without a processor](#create-a-tasklet-without-a-processor) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Use a callback](#use-a-callback) +- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Use a callback](#use-a-callback) + - [Java](#java-3) + - [Kotlin](#kotlin-3) A chunk-oriented step in Spring Batch consists of `ItemReader`, `ItemProcessor`, and `ItemWriter`, which are usually defined separately and then assembled to define a step. However, there are some issues with this approach: it is difficult to share data between `ItemReader`, `ItemProcessor`, and `ItemWriter`, and you need to see each respective file to understand the batch flow. Also, if the classes are not reused, they can make the elements of a job less coherent. @@ -298,9 +301,147 @@ open class TestJobConfig( } ``` +## Create a tasklet without a writer + +If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamSimpleReaderWriter` to define `ItemStreamReader` and `ItemProcessor` in a single class. + +### Java + +In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamSimpleReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @Override + public Integer read() { + if (count < totalCount) { + return count++; + } else { + return null; + } + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +You can statically import the method of `AdapterFactory` for better readability. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamSimpleReaderProcessor { + private var count = 0 + + override fun read(): Int? { + return if (count < totalCount) { + count++ + } else { + null + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Use a callback -You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamSimpleReaderProcessorWriter` and `ItemStreamSimpleReaderWriter`. You can selectively define a callback method. +You can define a callback method for `ItemStream`. You can selectively define a callback method. ### Java diff --git a/doc/ko/step/item-stream-flux-reader-processor-writer.md b/doc/ko/step/item-stream-flux-reader-processor-writer.md index 8f726535..5cbfc89b 100644 --- a/doc/ko/step/item-stream-flux-reader-processor-writer.md +++ b/doc/ko/step/item-stream-flux-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Processor 없이 Tasklet 작성하기](#processor-없이-tasklet-작성하기) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Callback 사용하기](#callback-사용하기) +- [Writer 없이 Tasklet 작성하기](#writer-없이-tasklet-작성하기) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Callback 사용하기](#callback-사용하기) + - [Java](#java-3) + - [Kotlin](#kotlin-3) Spring 진영에서는 Reactive library로 [Reactor](https://projectreactor.io/)를 사용하고 있습니다. Reactor에서는 여러 데이터에 대한 stream을 `Flux`로 제공합니다. `Flux`로 읽은 데이터를 Spring Batch의 `ItemReader`에서 사용하기 위해서는 `Flux`에서 단일 Item씩 뽑아내서 리턴하는 작업이 필요합니다. @@ -326,9 +329,156 @@ open class TestJobConfig( } ``` +## Writer 없이 Tasklet 작성하기 + +`ItemStreamReader` 와 `ItemProcessor` 만 묶어서 사용하고 싶은 경우 `ItemStreamFluxReaderProcessor`를 상속하여 단일 class에서 `ItemStreamReader`, `ItemProcessor`를 정의할 수 있습니다. + +### Java + +Java의 경우 `AdapterFactory`를 이용해서 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 변환하여 사용할 수 있습니다. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamFluxReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Flux readFlux(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return Flux.generate(sink -> { + if (count < totalCount) { + sink.next(count); + ++count; + } else { + sink.complete(); + } + }); + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +이 경우 `AdapterFactory`의 method를 static import를 해서 사용하는게 미관상 보기 더 좋습니다. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +Kotlin 사용시에는 Spring Batch Plus가 제공하는 extension function 을 사용하여 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 편리하게 변환할 수 있습니다. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamFluxReaderProcessor { + private var count = 0 + + override fun readFlux(executionContext: ExecutionContext): Flux { + println("totalCount: $totalCount") + return Flux.generate { sink -> + if (count < totalCount) { + sink.next(count) + ++count + } else { + sink.complete() + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Callback 사용하기 -`ItemStreamFluxReaderProcessorWriter`, `ItemStreamFluxReaderWriter` 에는 `ItemStreamReader`, `ItemStreamWriter`의 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. +각 Adapter 에는 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. ### Java diff --git a/doc/ko/step/item-stream-iterable-reader-processor-writer.md b/doc/ko/step/item-stream-iterable-reader-processor-writer.md index b694eccf..1af416e9 100644 --- a/doc/ko/step/item-stream-iterable-reader-processor-writer.md +++ b/doc/ko/step/item-stream-iterable-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Processor 없이 Tasklet 작성하기](#processor-없이-tasklet-작성하기) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Callback 사용하기](#callback-사용하기) +- [Writer 없이 Tasklet 작성하기](#writer-없이-tasklet-작성하기) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Callback 사용하기](#callback-사용하기) + - [Java](#java-3) + - [Kotlin](#kotlin-3) Spring Batch의 Chunk-oriented Step은 `ItemReader`, `ItemProcessor`, `ItemWriter`로 구성됩니다. Spring Batch에서는 일반적으로 `ItemReader`, `ItemProcessor`, `ItemWriter`를 각각 정의하고 이를 Step을 정의할 때 조립해서 사용합니다. 그런데 이 경우 `ItemReader`, `ItemProcessor`, `ItemWriter`간에 데이터 공유가 힘들고 배치의 흐름을 알기 위해서는 `ItemReader`, `ItemProcessor`, `ItemWriter`파일 각각을 살펴봐야 한다는 문제점이 있습니다. 또한 해당 클래스들이 재활용 되지 않는 케이스라면 Job의 응집도를 해치는 요소가 될 수 있습니다. @@ -328,9 +331,162 @@ open class TestJobConfig( } ``` +## Writer 없이 Tasklet 작성하기 + +`ItemStreamReader` 와 `ItemProcessor` 만 묶어서 사용하고 싶은 경우 `ItemStreamIterableReaderProcessor`를 상속하여 단일 class에서 `ItemStreamReader`, `ItemProcessor`를 정의할 수 있습니다. + +### Java + +Java의 경우 `AdapterFactory`를 이용해서 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 변환하여 사용할 수 있습니다. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamIterableReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Iterable readIterable(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return () -> new Iterator<>() { + @Override + public boolean hasNext() { + return count < totalCount; + } + + @Override + public Integer next() { + return count++; + } + }; + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +이 경우 `AdapterFactory`의 method를 static import를 해서 사용하는게 미관상 보기 더 좋습니다. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +Kotlin 사용시에는 Spring Batch Plus가 제공하는 extension function 을 사용하여 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 편리하게 변환할 수 있습니다. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamIterableReaderProcessor { + private var count = 0 + + override fun readIterable(executionContext: ExecutionContext): Iterable { + println("totalCount: $totalCount") + return Iterable { + object : Iterator { + override fun hasNext(): Boolean { + return count < totalCount + } + + override fun next(): Int { + return count++ + } + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Callback 사용하기 -`ItemStreamIterableReaderProcessorWriter`, `ItemStreamIterableReaderWriter` 에는 `ItemStreamReader`, `ItemStreamWriter`의 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. +각 Adapter 에는 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. ### Java diff --git a/doc/ko/step/item-stream-iterator-reader-processor-writer.md b/doc/ko/step/item-stream-iterator-reader-processor-writer.md index 404a3b99..91078f11 100644 --- a/doc/ko/step/item-stream-iterator-reader-processor-writer.md +++ b/doc/ko/step/item-stream-iterator-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Processor 없이 Tasklet 작성하기](#processor-없이-tasklet-작성하기) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Callback 사용하기](#callback-사용하기) +- [Writer 없이 Tasklet 작성하기](#writer-없이-tasklet-작성하기) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Callback 사용하기](#callback-사용하기) + - [Java](#java-3) + - [Kotlin](#kotlin-3) Spring Batch의 Chunk-oriented Step은 `ItemReader`, `ItemProcessor`, `ItemWriter`로 구성됩니다. Spring Batch에서는 일반적으로 `ItemReader`, `ItemProcessor`, `ItemWriter`를 각각 정의하고 이를 Step을 정의할 때 조립해서 사용합니다. 그런데 이 경우 `ItemReader`, `ItemProcessor`, `ItemWriter`간에 데이터 공유가 힘들고 배치의 흐름을 알기 위해서는 `ItemReader`, `ItemProcessor`, `ItemWriter`파일 각각을 살펴봐야 한다는 문제점이 있습니다. 또한 해당 클래스들이 재활용 되지 않는 케이스라면 Job의 응집도를 해치는 요소가 될 수 있습니다. @@ -324,9 +327,160 @@ open class TestJobConfig( } ``` +## Writer 없이 Tasklet 작성하기 + +`ItemStreamReader` 와 `ItemProcessor` 만 묶어서 사용하고 싶은 경우 `ItemStreamIteratorReaderProcessor`를 상속하여 단일 class에서 `ItemStreamReader`, `ItemProcessor`를 정의할 수 있습니다. + +### Java + +Java의 경우 `AdapterFactory`를 이용해서 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 변환하여 사용할 수 있습니다. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamIteratorReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @NonNull + @Override + public Iterator readIterator(@NonNull ExecutionContext executionContext) { + System.out.println("totalCount: " + totalCount); + return new Iterator<>() { + @Override + public boolean hasNext() { + return count < totalCount; + } + + @Override + public Integer next() { + return count++; + } + }; + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +이 경우 `AdapterFactory`의 method를 static import를 해서 사용하는게 미관상 보기 더 좋습니다. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +Kotlin 사용시에는 Spring Batch Plus가 제공하는 extension function 을 사용하여 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 편리하게 변환할 수 있습니다. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamIteratorReaderProcessor { + private var count = 0 + + override fun readIterator(executionContext: ExecutionContext): Iterator { + println("totalCount: $totalCount") + return object : Iterator { + override fun hasNext(): Boolean { + return count < totalCount + } + + override fun next(): Int { + return count++ + } + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Callback 사용하기 -`ItemStreamIteratorReaderProcessorWriter`, `ItemStreamIteratorReaderWriter` 에는 `ItemStreamReader`, `ItemStreamWriter`의 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. +각 Adapter 에는 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. ### Java diff --git a/doc/ko/step/item-stream-simple-reader-processor-writer.md b/doc/ko/step/item-stream-simple-reader-processor-writer.md index 6620026a..5ec1a62e 100644 --- a/doc/ko/step/item-stream-simple-reader-processor-writer.md +++ b/doc/ko/step/item-stream-simple-reader-processor-writer.md @@ -6,9 +6,12 @@ - [Processor 없이 Tasklet 작성하기](#processor-없이-tasklet-작성하기) - [Java](#java-1) - [Kotlin](#kotlin-1) -- [Callback 사용하기](#callback-사용하기) +- [Writer 없이 Tasklet 작성하기](#writer-없이-tasklet-작성하기) - [Java](#java-2) - [Kotlin](#kotlin-2) +- [Callback 사용하기](#callback-사용하기) + - [Java](#java-3) + - [Kotlin](#kotlin-3) Spring Batch의 Chunk-oriented Step은 `ItemReader`, `ItemProcessor`, `ItemWriter`로 구성됩니다. Spring Batch에서는 일반적으로 `ItemReader`, `ItemProcessor`, `ItemWriter`를 각각 정의하고 이를 Step을 정의할 때 조립해서 사용합니다. 그런데 이 경우 `ItemReader`, `ItemProcessor`, `ItemWriter`간에 데이터 공유가 힘들고 배치의 흐름을 알기 위해서는 `ItemReader`, `ItemProcessor`, `ItemWriter`파일 각각을 살펴봐야 한다는 문제점이 있습니다. 또한 해당 클래스들이 재활용 되지 않는 케이스라면 Job의 응집도를 해치는 요소가 될 수 있습니다. @@ -298,9 +301,147 @@ open class TestJobConfig( } ``` +## Writer 없이 Tasklet 작성하기 + +`ItemStreamReader` 와 `ItemProcessor` 만 묶어서 사용하고 싶은 경우 `ItemStreamSimpleReaderProcessor`를 상속하여 단일 class에서 `ItemStreamReader`, `ItemProcessor`를 정의할 수 있습니다. + +### Java + +Java의 경우 `AdapterFactory`를 이용해서 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 변환하여 사용할 수 있습니다. + +```java +@Component +@StepScope +class SampleTasklet implements ItemStreamSimpleReaderProcessor { + + @Value("#{jobParameters['totalCount']}") + private long totalCount; + + private int count = 0; + + @Override + public Integer read() { + if (count < totalCount) { + return count++; + } else { + return null; + } + } + + @Override + public String process(@NonNull Integer item) { + return "'" + item.toString() + "'"; + } +} +``` + +```java +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(AdapterFactory.itemStreamReader(sampleTasklet)) + .processor(AdapterFactory.itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +이 경우 `AdapterFactory`의 method를 static import를 해서 사용하는게 미관상 보기 더 좋습니다. + +```java +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader; +import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter; + +... + +@Configuration +public class TestJobConfig { + + @Bean + public Job testJob( + SampleTasklet sampleTasklet, + JobRepository jobRepository, + PlatformTransactionManager transactionManager + ) { + return new JobBuilder("testJob", jobRepository) + .start( + new StepBuilder("testStep", jobRepository) + .chunk(3, transactionManager) + .reader(itemStreamReader(sampleTasklet)) + .processor(itemProcessor(sampleTasklet)) + .writer(chunk -> System.out.println(chunk.getItems())) + .build() + ) + .build(); + } +} +``` + +### Kotlin + +Kotlin 사용시에는 Spring Batch Plus가 제공하는 extension function 을 사용하여 정의한 Tasklet을 `ItemStreamReader`, `ItemProcessor`로 편리하게 변환할 수 있습니다. + +```Kotlin +@Component +@StepScope +open class SampleTasklet( + @Value("#{jobParameters['totalCount']}") private var totalCount: Long, +) : ItemStreamSimpleReaderProcessor { + private var count = 0 + + override fun read(): Int? { + return if (count < totalCount) { + count++ + } else { + null + } + } + + override fun process(item: Int): String? { + return "'$item'" + } +} +``` + +```Kotlin +@Configuration +open class TestJobConfig( + private val batch: BatchDsl, + private val transactionManager: PlatformTransactionManager, +) { + @Bean + open fun testJob( + sampleTasklet: SampleTasklet, + ): Job = batch { + job("testJob") { + step("testStep") { + chunk(3, transactionManager) { + reader(sampleTasklet.asItemStreamReader()) + processor(sampleTasklet.asItemProcessor()) + writer { chunk -> println(chunk.items) } + } + } + } + } +} +``` + ## Callback 사용하기 -`ItemStreamSimpleReaderProcessorWriter`, `ItemStreamSimpleReaderWriter` 에는 `ItemStreamReader`, `ItemStreamWriter`의 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. +각 Adapter 에는 `ItemStream`에 대한 callback method도 같이 정의할 수 있습니다. Callback method는 선택적으로 정의할 수 있습니다. ### Java