ここでは、partitionによる並行処理を見て行きます。
自由度が高いのは、splitによる並行処理ですが、スケーリングを自動で行ってくれるのがpartitionの特徴です。
スケーリングとは、処理の量や種類などによって、スレッド数を多くしたり、少なくしたりすることです。
こうすることで、リソースをうまく使うことができます。
詳しくは、前の章で記述しています、並行処理の概要の記事を読んでみてください。
まず、以下のサンプルの目標（ゴール）を示します。
複数のファイルを同時に処理することを考えます。
処理内容は同じなのですが、ファイルが複数に分かれているようなケースを想像いただければと思います。
複数に分かれたファイルを順番に処理すると時間がかかります。
最近のPCはデュアルコアになっていたり、同時処理に向いたものが多くなってきています。
そのため、シングルに処理をするよりも、同時に複数処理をするほうが早くなるケースが多くなってきていると思います。
さて、ここでは、Spring Batchに載っていたサンプルを基にします。
【処理内容】
3つのCSVファイルを3つのスレッドを立てて、同時に処理します。
内容は、元のファイルを読み込んで、フィールドの順番を替えて出力することとします。
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<import resource="classpath:/remotetest/back-context.xml"/>
<job id="partitionJob" xmlns="http://www.springframework.org/schema/batch" incrementer="jobParametersIncrementer">
<step id="step">
<partition step="step01" partitioner="partitioner">
<handler grid-size="3" task-executor="taskExecutor" />
</partition>
</step>
</job>
<!-- enables the functionality of JobOperator.startNextInstance(jobName) -->
<bean id="jobParametersIncrementer" class="org.springframework.batch.core.launch.support.RunIdIncrementer" />
<!-- ファイルを読み込むパーティショナー -->
<bean id="partitioner" class="org.springframework.batch.core.partition.support.MultiResourcePartitioner">
<property name="resources" value="classpath:/remotetest/delimited*.txt" />
</bean>
<!-- 非同期タスクExecutor -->
<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />
<!-- 各スレッドで実行するステップ -->
<step id="step01" xmlns="http://www.springframework.org/schema/batch">
<tasklet transaction-manager="transactionManager">
<chunk writer="itemWriter" reader="itemReader" commit-interval="1" />
<listeners>
<listener ref="outputFileListener"/>
</listeners>
</tasklet>
</step>
<!-- 出力ファイル名をStepExecutionContextに設定するリスナー -->
<bean id="outputFileListener" class="remotetest.OutputFileListener"/>
<bean id="itemReader" scope="step" autowire-candidate="false" parent="itemReaderParent">
<property name="resource" value="#{stepExecutionContext[fileName]}" />
</bean>
<bean id="itemReaderParent" class="org.springframework.batch.item.file.FlatFileItemReader" abstract="true">
<property name="lineMapper">
<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<property name="lineTokenizer">
<bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<property name="delimiter" value="," />
<property name="names" value="name,credit,k" />
</bean>
</property>
<property name="fieldSetMapper">
<bean class="org.springframework.batch.item.file.mapping.PassThroughFieldSetMapper" />
</property>
</bean>
</property>
</bean>
<bean id="itemWriter" class="org.springframework.batch.item.file.FlatFileItemWriter" scope="step">
<property name="resource" value="file:c:/temp/#{stepExecutionContext[outFile]}" />
<property name="lineAggregator">
<bean class="org.springframework.batch.item.file.transform.DelimitedLineAggregator">
<property name="delimiter" value="," />
<property name="fieldExtractor">
<bean class="org.springframework.batch.item.file.transform.PassThroughFieldExtractor" />
</property>
</bean>
</property>
</bean>
</beans>
【タグについて】
partitionerタグは、内部にhandlerタグを持ちます。
handlerは、PartitionHandlerクラスに相当するものです。
handler内のgrid-sizeとは、生成するスレッドの最大数を設定します。
今回は3つにしますので、3と設定しています。
【Partionerについて】
さて、partitionerに設定しているPartitionerクラスを見てみてください。
自作しても良いのですが、既にSpringBatchが用意してくれているMultiResourcePartitionerクラスを使用しました。
MultiResourcePartitionerクラスは、*でファイルパスを指定できますので、複数のファイルを取得します。
ファイル名を取得したら、それをStepExecutionContextをファイルの数だけ生成し、
"fileName"というキー名でファイル名を登録します。
これにより、itemRaderのresourceに、#{stepExecutionContext[fileName]}という遅延バインディングでファイル名を設定できるように
なります。
出力ファイル名は、partitionerでは生成されませんので、outputFileListenerリスナーで設定しています。
これは既存では用意されていないので、自作することになります。
以下でそのサンプルも見てみることします。
【Partitionerが生成したStepExecutionContextがgrid-sizeより少ない場合】
Partitionerは、grid-sizeの3ファイルより少ない数（例えば2ファイル）を取得するかもしれません。
その場合は、PartitionHandlerが、スレッド数は2に自動で縮退します。
public class OutputFileListener implements StepExecutionListener {
@Override
public ExitStatus afterStep(StepExecution arg0) {
return null;
}
@Override
public void beforeStep(StepExecution se) {
String path = se.getExecutionContext().getString("fileName");
String file = StringUtils.getFilename(path);
se.getExecutionContext().put("outFile", file);
}
}
このリスナーは、partitionerが作った読み込みファイルパスから、ファイル名のみを取り出して、
StepExecutionContextの、"outFile"に設定します。
あとは、writerのリソースで、遅延バインディングで取り出します。
12,www,dddjjd
11,hhh,eee
00,fff,sss
上記と同様なカンマ区切りファイルをあと2ファイル用意して下さい。
用意できれば、後はジョブを実行していただければ、動作を確認できるはずです。
partitionerを使用すると、簡単にスケーリングを含めた処理を実現できます。
ただし、既存で用意されているクラスは少ないので、基本的には、上記で見たサンプルのように自作することが多そうです。
soracaneでは、Partitionerも提供予定です。
ユーザガイドやjavadocをご覧ください。
Created Date： 2010/01/12