春天批处理作业从多个来源读取 [英] Spring batch Job read from multiple sources

问题描述

如何从多个数据库中读取项目？我已经知道可以从文件。

以下示例适用于从多个文件读取

 。 .. 
< job id =readMultiFileJobxmlns =http://www.springframework.org/schema/batch> 
< step id =step1> 
< tasklet> 
< chunk reader =multiResourceReaderwriter =flatFileItemWriter
 commit-interval =1/> 
< / tasklet> 
< / step> 
< / job> 
 ... 
< bean id =multiResourceReader
 class =org.springframework.batch.item.file.MultiResourceItemReader> 
< property name =resourcesvalue =file：csv / inputs / domain  -  *。csv/> 
< property name =delegateref =flatFileItemReader/> 
< / bean> 
 ... 
  

---

这。

 < bean id =database2class =org.springframework.batch.item.database.JdbcCursorItemReader> ; 
< property name =namevalue =database2Reader/> 
< property name =dataSourceref =dataSource2/> 
< property name =sqlvalue =select image from object where'％/ images /％'/> 
< property name =rowMapper> 
< bean class =sym.batch.ImagesRowMapper2/> 
< / property> 
< / bean> 
  

解决方案

执行你所要求的;唯一的解决方案是写一个自定义的 ItemReader<> ，它委托 JdbcCursorItemReader （或 HibernateCursorItemReader 或任何通用的 ItemReader 实现）。

您需要准备所有必要的东西（数据源，真实数据库读取器），并将所有委派的读者绑定到您的自定义读取器。

编辑：
您需要使用<

  c $ c> ItemReader.read（）和mantain reader和委托状态。 class MyItemReader< T> implements ItemReader< T>，ItemStream {
 private ItemReader [] delegates; 
 private int delegateIndex; 
 private ItemReader< T> currentDelegate; 
 private ExecutionContext stepExecutionContext; 
 
 public void setDelegates（ItemReader [] delegates）{
 this.delegates = delegates; 
} 
 
 @BeforeStep 
 private void beforeStep（StepExecution stepExecution）{
 this.stepExecutionContext = stepExecution.getExecutionContext（）; 
} 
 
 public T read（）{
 T item = null; 
 if（null！= currentDelegate）{
 item = currentDelegate.read（）; 
 if（null == item）{
（（ItemStream）this.currentDelegate）.close（）; 
 this.currentDelegate = null; 
} 
} 
 //如果先前用尽，移动到下一个代理！ 
 if（null == item&& this.delegateIndex< this.delegates.length）{
 this.currentDelegate = this.delegates [this.currentIndex ++]; 
（（ItemStream）this.currentDelegate）.open（this.stepExecutionContext）; 
 update（this.stepExecutionContext）; 
 // Recurse to read（）通过代理模拟循环
 item = read（）; 
} 
 return item; 
} 
 
 public void open（ExecutionContext ctx）{
 //在打开还原期间活动读写器并恢复其状态
 if（ctx.containsKey（index ））{
 this.delegateIndex = ctx.getInt（index）; 
 this.currentDelegate = this.delegates [this.delegateIndex]; 
（（ItemStream）this.currentDelegate）.open（ctx）; 
} 
} 
 
 public void update（ExecutionContext ctx）{
 //更新当前委托索引和状态
 ctx.putInt（index this.delegateIndex）; 
 if（null！= this.currentDelegate）{
（（ItemStream）this.currentDelegate）.update（ctx）; 
} 
} 
 
 public void close（ExecutionContext ctx）{
 if（null！= this.currentDelegate）{
（（ItemStream）this。 currentDelegate）.close（）; 
} 
} 
  

---

 < bean id =myItemReaderclass = path.to.MyItemReader> 
< property name =delegates> 
< array> 
< ref bean =itemReader1/> 
< ref bean =itemReader2/> 
< ref bean =itemReader3/> 
< / array> 
< / property> 
< / bean> 
  
 
 
 
 
 
 
 EDIT2：请记住设置属性 name ;这是NECESSARY让MyItemReader.read（）正常工作
 < bean id =itemReader1class =JdbcCursorItemReader> ; 
< property name =namevalue =itemReader1/> 
<！ - 设置其他属性 - > 
< / bean> 
  
 
How can I read items from multiples databases? I already know that is possible from files.

the following example works for read from multiples files
...
<job id="readMultiFileJob" xmlns="http://www.springframework.org/schema/batch">
    <step id="step1">
    <tasklet>
        <chunk reader="multiResourceReader" writer="flatFileItemWriter"
            commit-interval="1" />
    </tasklet>
    </step>
</job>
...
<bean id="multiResourceReader"
    class=" org.springframework.batch.item.file.MultiResourceItemReader">
    <property name="resources" value="file:csv/inputs/domain-*.csv" />
    <property name="delegate" ref="flatFileItemReader" />
</bean>
...




three beans like this.
<bean id="database2" class="org.springframework.batch.item.database.JdbcCursorItemReader">
    <property name="name" value="database2Reader" />
    <property name="dataSource" ref="dataSource2" />
    <property name="sql" value="select image from object where image like '%/images/%'" />
    <property name="rowMapper">
        <bean class="sym.batch.ImagesRowMapper2" />
    </property>
</bean>

解决方案
There isn't a ready-to-use component that perform what you ask; the only solution is to write a custom ItemReader<> that delegates to JdbcCursorItemReader (or to HibernateCursorItemReader or to any generic ItemReader implementation).

You need to prepare all necessary stuff (datasource, session, real database readers) and bind all delegated readers to your custom reader.


EDIT:
You need to simulate a loop using recusion of ItemReader.read() and mantain reader and delegates state across job restarts.
class MyItemReader<T> implements ItemReader<T>, ItemStream {
  private ItemReader[] delegates;
  private int delegateIndex;
  private ItemReader<T> currentDelegate;
  private ExecutionContext stepExecutionContext;

  public void setDelegates(ItemReader[] delegates) {
    this.delegates = delegates;
  }

  @BeforeStep
  private void beforeStep(StepExecution stepExecution) {
    this.stepExecutionContext = stepExecution.getExecutionContext();
  }

  public T read() {
    T item = null;
    if(null != currentDelegate) {
      item = currentDelegate.read();
      if(null == item) {
        ((ItemStream)this.currentDelegate).close();
        this.currentDelegate = null;
      }
    }
    // Move to next delegate if previous was exhausted!
    if(null == item && this.delegateIndex< this.delegates.length) {
      this.currentDelegate = this.delegates[this.currentIndex++];
      ((ItemStream)this.currentDelegate).open(this.stepExecutionContext);
      update(this.stepExecutionContext);
      // Recurse to read() to simulate loop through delegates
      item = read();
    }
    return item;
  }

  public void open(ExecutionContext ctx) {
    // During open restore last active reader and restore its state
    if(ctx.containsKey("index")) {
      this.delegateIndex = ctx.getInt("index");
      this.currentDelegate = this.delegates[this.delegateIndex];
      ((ItemStream)this.currentDelegate ).open(ctx);
    }
  }

  public void update(ExecutionContext ctx) {
    // Update current delegate index and state
    ctx.putInt("index", this.delegateIndex);
    if(null != this.currentDelegate) {
      ((ItemStream)this.currentDelegate).update(ctx);
    }
  }

  public void close(ExecutionContext ctx) {
    if(null != this.currentDelegate) {
      ((ItemStream)this.currentDelegate).close();
  }
}




<bean id="myItemReader" class=path.to.MyItemReader>
  <property name="delegates">
    <array>
      <ref bean="itemReader1"/>
      <ref bean="itemReader2"/>
      <ref bean="itemReader3"/>
    </array>
  </property>
</bean>




EDIT2: Remember to set property name; this is NECESSARY to let MyItemReader.read() works correctly
<bean id="itemReader1" class="JdbcCursorItemReader">
  <property name="name" value="itemReader1" />
  <!-- Set other properties -->
</bean>


                        
这篇关于春天批处理作业从多个来源读取的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！