Spring批處理從多個來源讀取的作業

Question

如何從多個數據庫中讀取項目？我已經知道這是可能的文件。
下面的示例適用於讀取從倍數文件

... 
<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> 
... 

---

三豆這樣的。

<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> 

Answer 1

沒有一個隨時可用的組件來執行你所要求的;唯一的解決方案是編寫一個代表JdbcCursorItemReader（或HibernateCursorItemReader或任何通用ItemReader實現）的自定義ItemReader<>。
您需要準備好所有必需的東西（數據源，會話，真實數據庫讀取器）並將所有委派的讀者綁定到您的自定義閱讀器。

編輯： 您需要模擬一個循環，使用recrecusion 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; 
     } 
    } 
    // 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：請記住，設置屬性名;這是必要讓MyItemReader.read（）正常工作

<bean id="itemReader1" class="JdbcCursorItemReader"> 
    <property name="name" value="itemReader1" /> 
    <!-- Set other properties --> 
</bean> 

Answer 2

我建議一個棘手的方法。如果我們假設你的mysql數據源的表是基數，並且該表中的每一行對應其他mysql數據源表的行（就像在不同數據源中的連接表一樣），那麼可以在批處理作業itemreader中執行此操作。這樣的例子;

Spring DataSource Configuration;

<bean id="mySqlDataSource1" class="org.apache.commons.dbcp.BasicDataSource"> 
    <property name="driverClassName" value="${database1.driverClassName}"/> 
    <property name="url" value="${database1.url}"/> 
    <property name="username" value="${database1.username}"/> 
    <property name="password" value="${database1.password}"/> 
    <property name="validationQuery" value="${database1.validationQuery}"/> 
</bean> 

<bean id="mySqlDataSource2" class="org.apache.commons.dbcp.BasicDataSource"> 
    <property name="driverClassName" value="${database2.driverClassName}"/> 
    <property name="url" value="${database2.url}"/> 
    <property name="username" value="${database2.username}"/> 
    <property name="password" value="${database2.password}"/> 
    <property name="validationQuery" value="${database2.validationQuery}"/> 
</bean> 

你批job.xml

<bean id="multiDatasorceReader" class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step"> 
    <property name="dataSource" ref="mySqlDataSource1" /> 
    <property name="rowMapper" ref="multiDatasourceRowMapper" /> 
    <property name="sql"> 
     <value> 
      SELECT * FROM xyz 
     </value> 
    </property> 
</bean> 

<bean id="multiDatasourceRowMapper" class="yourpackage.MultiDatasourceRowMapper" scope="step"> 
    <property name="secondDataSource" ref="mySqlDataSource2" /> 
    <property name="secondSql"> 
     <value> 
      SELECT * FROM abc 
     </value> 
    </property> 
</bean> 

你RowMapper的樣子;

public class MultiDatasourceRowMapper implements RowMapper<String> { 

    private DataSource secondDataSource; 

    private String secondSql; 

    public String mapRow(ResultSet rs, int arg1) throws SQLException { 
     Connection conn = secondDataSource.getConnection(); 
     PreparedStatement prep = conn.prepareStatement(secondSql); 

     // Do Something 

     return ""; 
    } 

    public void setSecondDataSource(DataSource secondDataSource) { 
     this.secondDataSource = secondDataSource; 
    } 

    public void setSecondSql(String secondSql) { 
     this.secondSql = secondSql; 
    } 

} 

Answer 3

我提出一個簡單的解決方法可能不適合所有的情況，但在許多有用：

簡單的定義：

• 2名讀者，每一個數據庫
• 2步
• 一個作業，其中包含兩個步驟

這兩個步驟幾乎完全相同，它們引用相同的處理器和作者，但他們有不同的讀者。他們將被連續調用。

此設置是否有效取決於處理器和寫入器（當它們在不同的步驟中調用時它們是否仍然正常工作）。就我而言，將appendAllowed=true設置爲寫入器就足夠了，這樣兩個步驟都可以寫入同一個文件。