知識庫 / Spring / Spring Boot RSS 訂閱

5. 實現處理器路由器

為了將我們的處理器連接到適當的記錄中，我們需要一個路由機制，該機制會檢查每個客户的類型字段：

public class CustomerProcessorRouter implements ItemProcessor<Customer, Customer> {
    private final TypeAProcessor typeAProcessor;
    private final TypeBProcessor typeBProcessor;

    public CustomerProcessorRouter(TypeAProcessor typeAProcessor, 
      TypeBProcessor typeBProcessor) {
        this.typeAProcessor = typeAProcessor;
        this.typeBProcessor = typeBProcessor;
    }

    @Override
    public Customer process(Customer customer) throws Exception {
        if ("A".equals(customer.getType())) {
            return typeAProcessor.process(customer);
        } else if ("B".equals(customer.getType())) {
            return typeBProcessor.process(customer);
        }
        return customer;
    }
}

我們的路由器類會檢查每個傳入的 Customer 對象，並根據 type 字段將其委派到適當的處理器。這種設計使我們的處理邏輯與 job 定義中的單一處理步驟保持分離，同時保持了邏輯的清晰性。

6. 配置多個寫入器

在根據類型對我們的數據進行不同的處理後，我們希望將結果寫入多個目標位置。我們將實現數據庫寫入器和平面文件寫入器。

6.1. 數據庫寫入器配置

我們首先創建數據庫寫入器組件，該組件將處理所有 JPA 操作：

@Bean
public JpaItemWriter<Customer> dbWriter(EntityManagerFactory entityManagerFactory) {
    JpaItemWriter<Customer> writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(entityManagerFactory);
    return writer;
}

這位作者使用 JPA 將我們處理過的 Customer 對象持久化到數據庫。在作業執行期間，這個 JpaItemWriter 將持久化我們處理過的 Customer 對象到配置的數據庫，處理所有必要的 JPA 操作，包括插入和更新。

6.2. 平文件寫入器配置

對於我們的次要輸出目的地，我們實現了平文件寫入器，該寫入器生成 CSV 文件：

@Bean
public FlatFileItemWriter<Customer> fileWriter() {
    return new FlatFileItemWriterBuilder<Customer>()
      .name("customerItemWriter")
      .resource(new FileSystemResource("output/processed_customers.txt"))
      .delimited()
      .delimiter(",")
      .names("id", "name", "email", "type")
      .build();
}

FlatFileItemWriter 配置為使用逗號作為分隔符，幷包含與我們的 Customer 實體屬性相匹配的顯式字段命名。 在作業執行期間，該寫入器將創建一個結構化的 CSV 文件，其中包含所有已處理的客户記錄，格式符合指定要求。

6.3. 組合寫入器（Composite Writer）

為了同時將數據寫入多個目標，我們將使用 Spring Batch 的 CompositeItemWriter：

@Bean
public CompositeItemWriter<Customer> compositeWriter(
  JpaItemWriter<Customer> dbWriter,
  FlatFileItemWriter<Customer> fileWriter) {
    CompositeItemWriter<Customer> writer = new CompositeItemWriter<>();
    writer.setDelegates(List.of(dbWriter, fileWriter));
    return writer;
}

這個複合寫入器作為寫入器的委託，確保每個處理過的項目都寫入所有目標位置。委託的順序決定了寫入的順序。

7. 配置步驟和作業

現在，讓我們將所有內容整合在一起，通過創建步驟和作業配置來實現：

@Bean
public Job processCustomersJob(JobBuilderFactory jobs,
  StepBuilderFactory steps,
  FlatFileItemReader<Customer> reader,
  CustomerProcessorRouter processor,
  CompositeItemWriter<Customer> writer) {

    Step step = steps.get("processCustomersStep")
      .<Customer, Customer>chunk(10)
      .reader(reader)
      .processor(processor)
      .writer(writer)
      .build();

    return jobs.get("customerProcessingJob")
      .start(step)
      .build();
}

此作業配置定義了一個單步操作，它以 10 個客户的塊讀取客户數據，通過我們的路由器處理每個客户，並將結果同時寫入數據庫和平面文件。

8. 運行和測試作業

為了驗證我們的批處理作業按預期運行，我們編寫一個集成測試，該測試啓動作業並斷言不同客户類型的數據庫和輸出文件結果：

List<Customer> dbCustomers = jdbcTemplate.query(
    "SELECT id, name, email, type FROM customer WHERE type = 'A'",
    (rs, rowNum) -> new Customer(
        rs.getLong("id"),
        rs.getString("name"),
        rs.getString("email"),
        rs.getString("type"))
);

assertFalse(dbCustomers.isEmpty());

dbCustomers.forEach(c -> {
    assertEquals(c.getName(), c.getName().toUpperCase());
    assertTrue(c.getEmail().startsWith("A_"));
});

Path outputFile = Paths.get("output/processed_customers.txt");
assertTrue(Files.exists(outputFile));

List<String> lines = Files.readAllLines(outputFile);

boolean hasTypeB = lines.stream().anyMatch(line -> line.endsWith(",B"));
assertTrue(hasTypeB);

lines.forEach(line -> {
    String[] parts = line.split(",");
    if ("B".equals(parts[3])) {
        assertEquals(parts[1], parts[1].toLowerCase());
        assertTrue(parts[2].startsWith("B_"));
    }
});

從測試用例來看，首先我們查詢數據庫以驗證類型為 A 的客户已成功保存，且他們的姓名已轉換為大寫，電子郵件前綴為“A_”。接下來，我們還讀取輸出文件以確認類型為 B 的客户，他們的姓名已轉換為小寫，電子郵件前綴為“B_”。

9. 結論

在本文中，我們學習瞭如何使用單個讀取器但多個處理器和寫入器配置一個 Spring Batch 任務。我們從 CSV 文件中讀取數據，根據其內容將每個記錄路由到特定的處理器，最後將寫入任務委派給多個寫入器。