在 Spring Batch 中,我尝试读取 CSV 文件,并希望将每一行分配给一个单独的线程并处理它。我尝试通过使用 TaskExecutor 来实现它,但是所有线程正在发生的事情是一次选择同一行。我也尝试使用 Partioner 来实现这个概念,也发生了同样的事情。请参阅下面我的配置 Xml。
步骤说明
<step id="Step2">
<tasklet task-executor="taskExecutor">
<chunk reader="reader" processor="processor" writer="writer" commit-interval="1" skip-limit="1">
</chunk>
</tasklet>
</step>
<bean id="reader" class="org.springframework.batch.item.file.FlatFileItemReader">
<property name="resource" value="file:cvs/user.csv" />
<property name="lineMapper">
<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
<!-- split it -->
<property name="lineTokenizer">
<bean
class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
<property name="names" value="userid,customerId,ssoId,flag1,flag2" />
</bean>
</property>
<property name="fieldSetMapper">
<!-- map to an object -->
<bean
class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
<property name="prototypeBeanName" value="user" />
</bean>
</property>
</bean>
</property>
</bean>
<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor">
<property name="concurrencyLimit" value="4"/>
我尝试过不同类型的任务执行器,但它们的行为方式都相同。如何将每一行分配给单独的线程?
FlatFileItemReader 不是线程安全的。在您的示例中,您可以尝试将 CSV 文件拆分为更小的 CSV 文件,然后使用多资源分区器 http://docs.spring.io/spring-batch/apidocs/org/springframework/batch/core/partition/support/MultiResourcePartitioner.html处理它们中的每一个。这可以分两步完成,一个用于分割原始文件(如 10 个较小的文件),另一个用于处理分割后的文件。这样您就不会遇到任何问题,因为每个文件将由一个线程处理。
Example:
<batch:job id="csvsplitandprocess">
<batch:step id="step1" next="step2master">
<batch:tasklet>
<batch:chunk reader="largecsvreader" writer="csvwriter" commit-interval="500">
</batch:chunk>
</batch:tasklet>
</batch:step>
<batch:step id="step2master">
<partition step="step2" partitioner="partitioner">
<handler grid-size="10" task-executor="taskExecutor"/>
</partition>
</batch:step>
</batch:job>
<batch:step id="step2">
<batch:tasklet>
<batch:chunk reader="smallcsvreader" writer="writer" commit-interval="100">
</batch:chunk>
</batch:tasklet>
</batch:step>
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="10" />
<property name="maxPoolSize" value="10" />
</bean>
<bean id="partitioner"
class="org.springframework.batch.core.partition.support.MultiResourcePartitioner">
<property name="resources" value="file:cvs/extracted/*.csv" />
</bean>
除了分区之外,另一种选择可能是自定义线程安全读取器,它将为每一行创建一个线程,但分区可能是您的最佳选择
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)