@Configuration
@EnableBatchProcessing
public class BatchConfig {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Bean
    public ItemReader<User> userItemReader() {
        // 從數(shù)據(jù)庫讀取數(shù)據(jù)
        return new JdbcCursorItemReaderBuilder<User>()
            .name("userItemReader")
            .dataSource(dataSource)
            .sql("SELECT id, name, email FROM users WHERE status = 'ACTIVE'")
            .rowMapper(new BeanPropertyRowMapper<>(User.class))
            .build();
    }
    @Bean
    public ItemProcessor<User, UserDTO> userItemProcessor() {
        return user -> {
            UserDTO dto = new UserDTO();
            dto.setId(user.getId());
            dto.setName(user.getName().toUpperCase());
            dto.setEmail(user.getEmail().toLowerCase());
            return dto;
        };
    }
    @Bean
    public ItemWriter<UserDTO> userItemWriter() {
        // 寫入到文件
        return items -> {
            for (UserDTO item : items) {
                System.out.println("Processing user: " + item.getName());
                // 寫入到文件或其他目標(biāo)
            }
        };
    }
    @Bean
    public Step processUserStep() {
        return stepBuilderFactory.get("processUserStep")
            .<User, UserDTO>chunk(10)
            .reader(userItemReader())
            .processor(userItemProcessor())
            .writer(userItemWriter())
            .build();
    }
    @Bean
    public Job processUserJob() {
        return jobBuilderFactory.get("processUserJob")
            .incrementer(new RunIdIncrementer())
            .flow(processUserStep())
            .end()
            .build();
    }
}

@Configuration
public class DataSourceConfig {
    @Bean
    public DataSource dataSource() {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mysql://localhost:3306/batch_db");
        config.setUsername("root");
        config.setPassword("password");
        config.setMaximumPoolSize(10);
        return new HikariDataSource(config);
    }
    @Bean
    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
        return new JdbcTemplate(dataSource);
    }
}

@Configuration
public class JobRepositoryConfig {
    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(dataSource);
        factory.setTransactionManager(transactionManager);
        factory.setIsolationLevelForCreate("ISOLATION_SERIALIZABLE");
        factory.setTablePrefix("BATCH_");
        factory.setMaxVarCharLength(1000);
        return factory.getObject();
    }
    @Bean
    public PlatformTransactionManager transactionManager(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    }
    @Bean
    public JobLauncher jobLauncher(JobRepository jobRepository) throws Exception {
        SimpleJobLauncher launcher = new SimpleJobLauncher();
        launcher.setJobRepository(jobRepository);
        launcher.setTaskExecutor(new SimpleAsyncTaskExecutor());
        return launcher;
    }
}

@Bean
public ItemReader<Customer> customerItemReader(DataSource dataSource) {
    return new JdbcPagingItemReaderBuilder<Customer>()
        .name("customerItemReader")
        .dataSource(dataSource)
        .selectClause("SELECT id, first_name, last_name, email, phone")
        .fromClause("FROM customers")
        .whereClause("WHERE last_updated > :lastUpdated")
        .parameterValues(Collections.singletonMap("lastUpdated", LocalDateTime.now().minusDays(1)))
        .sortKeys(Collections.singletonMap("id", Order.ASCENDING))
        .rowMapper(new BeanPropertyRowMapper<>(Customer.class))
        .pageSize(100)
        .build();
}

@Bean
public ItemReader<Product> productItemReader() {
    return new FlatFileItemReaderBuilder<Product>()
        .name("productItemReader")
        .resource(new ClassPathResource("products.csv"))
        .delimited()
        .names("id", "name", "price", "quantity")
        .fieldSetMapper(fieldSet -> {
            Product product = new Product();
            product.setId(fieldSet.readLong("id"));
            product.setName(fieldSet.readString("name"));
            product.setPrice(fieldSet.readBigDecimal("price"));
            product.setQuantity(fieldSet.readInt("quantity"));
            return product;
        })
        .build();
}

public class CustomItemReader implements ItemReader<String> {
    private final List<String> items;
    private int index = 0;
    public CustomItemReader(List<String> items) {
        this.items = items;
    }
    @Override
    public String read() {
        if (index < items.size()) {
            return items.get(index++);
        }
        return null;
    }
}
@Bean
public ItemReader<String> customItemReader() {
    List<String> items = Arrays.asList("item1", "item2", "item3", "item4", "item5");
    return new CustomItemReader(items);
}

public class ProductProcessor implements ItemProcessor<Product, ProductDTO> {
    @Override
    public ProductDTO process(Product item) {
        ProductDTO dto = new ProductDTO();
        dto.setId(item.getId());
        dto.setName(item.getName());
        dto.setPrice(item.getPrice());
        dto.setQuantity(item.getQuantity());
        dto.setTotalValue(item.getPrice().multiply(BigDecimal.valueOf(item.getQuantity())));
        return dto;
    }
}
@Bean
public ItemProcessor<Product, ProductDTO> productProcessor() {
    return new ProductProcessor();
}

public class OrderProcessor implements ItemProcessor<Order, Order> {
    @Override
    public Order process(Order item) {
        if (item.getStatus().equals(OrderStatus.PENDING)) {
            item.setStatus(OrderStatus.PROCESSED);
            item.setProcessedAt(LocalDateTime.now());
            return item;
        }
        return null; // 跳過非待處理訂單
    }
}
@Bean
public ItemProcessor<Order, Order> orderProcessor() {
    return new OrderProcessor();
}

public class CompositeItemProcessor<T, R> implements ItemProcessor<T, R> {
    private final List<ItemProcessor> processors;
    public CompositeItemProcessor(List<ItemProcessor> processors) {
        this.processors = processors;
    }
    @Override
    public R process(T item) {
        Object result = item;
        for (ItemProcessor processor : processors) {
            result = processor.process(result);
            if (result == null) {
                return null;
            }
        }
        return (R) result;
    }
}
@Bean
public ItemProcessor<Customer, CustomerDTO> customerProcessor() {
    List<ItemProcessor> processors = new ArrayList<>();
    processors.add(new ValidationProcessor());
    processors.add(new TransformationProcessor());
    processors.add(new EnrichmentProcessor());
    return new CompositeItemProcessor<>(processors);
}

@Bean
public ItemWriter<CustomerDTO> customerItemWriter(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<CustomerDTO>()
        .dataSource(dataSource)
        .sql("INSERT INTO customer_processed (id, first_name, last_name, email, processed_at) VALUES (:id, :firstName, :lastName, :email, :processedAt)")
        .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
        .build();
}

@Bean
public ItemWriter<ProductDTO> productItemWriter() {
    return new FlatFileItemWriterBuilder<ProductDTO>()
        .name("productItemWriter")
        .resource(new FileSystemResource("output/products-processed.csv"))
        .delimited()
        .names("id", "name", "price", "quantity", "totalValue")
        .headerCallback(writer -> writer.write("ID,Name,Price,Quantity,Total Value"))
        .build();
}

public class CustomItemWriter implements ItemWriter<User> {
    private final Logger logger = LoggerFactory.getLogger(CustomItemWriter.class);
    @Override
    public void write(List<? extends User> items) {
        for (User item : items) {
            logger.info("Writing user: {}", item.getName());
            // 寫入到外部系統(tǒng)或其他目標(biāo)
        }
    }
}
@Bean
public ItemWriter<User> customItemWriter() {
    return new CustomItemWriter();
}

@Service
public class BatchService {
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private Job processUserJob;
    public void runProcessUserJob() throws Exception {
        JobParameters jobParameters = new JobParametersBuilder()
            .addString("jobName", "processUserJob")
            .addLong("time", System.currentTimeMillis())
            .toJobParameters();
        JobExecution execution = jobLauncher.run(processUserJob, jobParameters);
        System.out.println("Job execution status: " + execution.getStatus());
    }
}

@RestController
@RequestMapping("/api/batch")
public class BatchController {
    @Autowired
    private JobExplorer jobExplorer;
    @GetMapping("/jobs")
    public List<JobInstance> getJobs() {
        return jobExplorer.getJobInstances("processUserJob", 0, 10);
    }
    @GetMapping("/executions/{jobInstanceId}")
    public List<JobExecution> getExecutions(@PathVariable Long jobInstanceId) {
        JobInstance jobInstance = jobExplorer.getJobInstance(jobInstanceId);
        return jobExplorer.getJobExecutions(jobInstance);
    }
    @GetMapping("/steps/{jobExecutionId}")
    public List<StepExecution> getSteps(@PathVariable Long jobExecutionId) {
        JobExecution jobExecution = jobExplorer.getJobExecution(jobExecutionId);
        return jobExecution.getStepExecutions();
    }
}

@Configuration
@EnableScheduling
public class BatchScheduler {
    @Autowired
    private JobLauncher jobLauncher;
    @Autowired
    private Job processUserJob;
    @Scheduled(cron = "0 0 0 * * ?") // 每天凌晨執(zhí)行
    public void runDailyJob() throws Exception {
        JobParameters jobParameters = new JobParametersBuilder()
            .addString("jobName", "processUserJob")
            .addLong("time", System.currentTimeMillis())
            .toJobParameters();
        jobLauncher.run(processUserJob, jobParameters);
    }
}

@Bean
public Step processOrderStep() {
    return stepBuilderFactory.get("processOrderStep")
        .<Order, Order>chunk(10)
        .reader(orderItemReader())
        .processor(orderItemProcessor())
        .writer(orderItemWriter())
        .faultTolerant()
        .skipLimit(10)
        .skip(OrderProcessingException.class)
        .retryLimit(3)
        .retry(ConnectionException.class)
        .build();
}