當前位置：首頁 > 科技 > 軟件

SpringBatch高階應用：大數據批處理框架實戰指南

來源：責編：時間：2024-05-07 09:12:51 163觀看

導讀本篇文章主要內容：通過Spring Batch從一個庫中讀取數據進過處理后寫入到另外一個庫中。
1. 環境準備1.1 引入依賴<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-batch

本篇文章主要內容：通過Spring Batch從一個庫中讀取數據進過處理后寫入到另外一個庫中。

1. 環境準備

1.1 引入依賴

<dependency>  <groupId>org.springframework.boot</groupId>  <artifactId>spring-boot-starter-batch</artifactId></dependency><dependency>  <groupId>org.springframework.boot</groupId>  <artifactId>spring-boot-starter-data-jpa</artifactId></dependency>

2.2 配置Job

配置Job啟動器

@BeanJobLauncher userJobLauncher(JobRepository userJobRepository) {  SimpleJobLauncher jobLauncher = new SimpleJobLauncher() ;  jobLauncher.setJobRepository(userJobRepository) ;  return jobLauncher ;}

配置任務Repository存儲元信息

@BeanJobRepository userJobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) {  JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean() ;  factory.setDatabaseType("mysql") ;  factory.setTransactionManager(transactionManager) ;  factory.setDataSource(dataSource) ;  try {    factory.afterPropertiesSet() ;     return factory.getObject() ;  } catch (Exception e) {    throw new RuntimeException(e) ;  }}

配置ItemReader讀取器

@BeanItemReader<User> userReader(JobOperator jobOperator) throws Exception {  JpaPagingItemReaderBuilder<User> builder = new JpaPagingItemReaderBuilder<>() ;  builder.entityManagerFactory(entityManagerFactory) ;  // 每次分頁查詢多少條數據  builder.pageSize(10) ;  builder.queryString("select u from User u where u.uid <= 50") ;  builder.saveState(true) ;  builder.name("userReader") ;  return builder.build() ;}

配置數據源，該數據源是用來寫入操作的

public DataSource dataSource() {  HikariDataSource dataSource = new HikariDataSource() ;  dataSource.setJdbcUrl("jdbc:mysql://localhost:3306/testjpa?serverTimezone=GMT%2B8&useSSL=false") ;  dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver") ;  dataSource.setUsername("root") ;  dataSource.setPassword("xxxooo") ;  return dataSource ;}

配置ItemWriter用來寫入操作（當前庫的數據寫入到另外一個庫，上面的數據源）

@BeanItemWriter<User> userWriter() {  // 通過JDBC批量處理  JdbcBatchItemWriterBuilder<User> builder = new JdbcBatchItemWriterBuilder<>() ;  DataSource dataSource = dataSource() ;  builder.dataSource(dataSource) ;  builder.sql("insert into st (id, name, sex, mobile, age, birthday) values (?, ?, ?, ?, ?, ?)") ;  builder.itemPreparedStatementSetter(new ItemPreparedStatementSetter<User>() {    @Override    public void setValues(User item, PreparedStatement ps) throws SQLException {      ps.setInt(1, item.getUid()) ;      ps.setString(2, item.getName()) ;      ps.setString(3, item.getSex()) ;      ps.setString(4, item.getMobile()) ;      ps.setInt(5, item.getAge()) ;      ps.setObject(6, item.getBirthday()) ;    }  }) ;  return builder.build() ;}

配置ItemProcessor處理器，數據從當前庫讀取處理后經過處理后再寫入另外的庫中

@BeanItemProcessor<User, User> userProcessor() {  return new ItemProcessor<User, User>() {    @Override    public User process(User item) throws Exception {      System.out.printf("%s - 開始處理數據：%s%n", Thread.currentThread().getName(), item.toString()) ;      // 模擬耗時操作      TimeUnit.SECONDS.sleep(1) ;      // 在這里你可以對數據進行相應的處理。      return item ;    }  } ;}

配置Step將ItemReader、ItemProcessor、ItemWriter串聯在一起。

@BeanStep userStep1(ItemReader<User> userReader, ItemProcessor<User, User> userProcessor, ItemWriter<User> userWriter) {  return steps.get("userStep1")    .<User, User>chunk(5)    .reader(userReader)    .processor(userProcessor)    .writer(userWriter)    .build() ;}

配置Job，Job是封裝整個批處理流程的實體。在 Spring Batch 中，Job只是Step實例的容器。它將邏輯上屬于一個流程的多個步驟組合在一起，并允許對所有步驟的全局屬性（如可重啟性）進行配置。作業配置包含：

簡單的工作名稱。
Step實例的定義和排序。
Job是否可重新啟動。

@BeanJob userJob(Step userStep1, Step userStep2) {  return jobs.get("userJob").start(userStep1).build();}

以上是Spring Batch定義配置一個Job所需的核心組件。接下來會以上面的基礎配置進行高階知識點進行介紹。

2. 高階配置管理

2.1 通過Controller接口啟動Job

@RequestMapping("/userJob")public class UserJobController {  @Resource  private JobLauncher userJobLauncher ;  @GetMapping("/start")  public Object start() throws Exception {    JobParameters jobParameters = new JobParameters() ;    this.userJobLauncher.run(userJob, jobParameters) ;    return "started" ;  }}

通過JobLauncher#run方法啟動Job。當你調用該接口時，你會發現接口一直不會返回，一直阻塞，下圖是Job的啟動序列

圖片

根據上圖能知道，當你調用run方法后，會等待整個Job退出狀態為FINISHED或者FAILED后才能結束。所以，你需要異步完成，以便 SimpleJobLauncher 立即返回給調用者。而正確的序列應該是如下：

圖片

上圖通過異步方式啟動Job序列。

2.2 異步啟動Job

@BeanTaskExecutor taskExecutor() {  ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor() ;  taskExecutor.setThreadNamePrefix("spring_batch_launcher") ;  taskExecutor.setCorePoolSize(10) ;  taskExecutor.setMaxPoolSize(10) ;  taskExecutor.initialize() ;   return taskExecutor ;}@BeanJobLauncher userJobLauncher(JobRepository userJobRepository) {  SimpleJobLauncher jobLauncher = new SimpleJobLauncher() ;  jobLauncher.setJobRepository(userJobRepository) ;  jobLauncher.setTaskExecutor(taskExecutor()) ;  return jobLauncher ;}

通過上面配置后，Job啟動將是異步的會直接返回JobExecution。

2.3 重啟Job

當一個Job正在執行，由于斷電或者強制終止了程序。當程序恢復后你希望能夠接著程序終止前的進度繼續執行，這時候你需要進行如下的操作（本人沒有發現有什么API能夠操作的，可能文檔沒看仔細）。

當程序非正常終止是，下面兩張表的狀態都是STARTED，END_TIME為null

batch_job_execution表

圖片

batch_step_execution表

圖片

想要重新啟動必須將上面的狀態修改為STOPPED，END_TIME字段設置上值（是什么值無所謂）。

然后我們就可以繼續使用上面的Controller接口啟動任務繼續執行了。

2.4 多線程執行Step

為了加快程序的執行，我們可以為Step配置線程池

@BeanStep userStep1(ItemReader<User> userReader, ItemProcessor<User, User> userProcessor, ItemWriter<User> userWriter) {  return steps.get("userStep1")    .<User, User>chunk(5)    .reader(userReader)    .processor(userProcessor)    .writer(userWriter)    // 配置線程池    .taskExecutor(taskExecutor())    .build() ;}

注意：Step中使用的任何池化資源（如數據源）都可能對并發性設置限制。請確保這些資源池至少與步驟中所需的并發線程數一樣大。

通過上面配置線程池后，你將在控制臺看到如下輸出。

圖片

默認將有4個線程同時進行處理。可以通過如下配置進行調整

@BeanStep userStep1(ItemReader<User> userReader, ItemProcessor<User, User> userProcessor, ItemWriter<User> userWriter) {  return steps.get("userStep1")      // ...      // 節流限制10，這里配置的大小應該與你的數據庫連接池大小及使用的線程池核心線程數一致。      .throttleLimit(10)      .build() ;}

2.5 重復啟動Job

要想重復啟動Job，我們可以在啟動Job時設置不同的JobParameters參數，只要參數不同那么就可以重復的啟動Job。如下示例：

@GetMapping("/start/{page}")public Object start(@PathVariable("page") Long page) throws Exception {  Map<String, JobParameter> parameters = new HashMap<>() ;  // 每次設置的參數值不同即可。  parameters.put("page", new JobParameter(page)) ;  JobParameters jobParameters = new JobParameters(parameters) ;  this.userJobLauncher.run(userJob, jobParameters) ;  return "started" ;}

以上是本篇文章的全部內容，希望對你有幫助。

本文鏈接：http://www.www897cc.com/showinfo-26-87012-0.htmlSpringBatch高階應用：大數據批處理框架實戰指南

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇：架構設計中如何應對接口級故障？

下一篇： Web Components 取代 Vue？我覺得不太行！

標簽：

熱門焦點

石頭自清潔掃拖機器人G10S評測：多年黑科技集大成之作懶人終極福音

科技圈經常能看到一個詞叫“縫合怪”，用來形容那些把好多功能或者外觀結合在一起的產品，通常這樣的詞是貶義詞，但如果真的是產品縫合的好、縫合的實用的話，那它就成了中性詞，今
0糖0卡0脂旭日森林仙草烏龍茶優惠：15瓶到手29元

旭日森林無糖仙草烏龍茶510ml*15瓶平時要賣為79.9元，今日下單領取50元優惠券，到手價為29.9元。產品規格：0糖0卡0脂，添加草本仙草汁，清涼爽口，富含茶多酚，保留
三言兩語說透柯里化和反柯里化

JavaScript中的柯里化(Currying)和反柯里化(Uncurrying)是兩種很有用的技術，可以幫助我們寫出更加優雅、泛用的函數。本文將首先介紹柯里化和反柯里化的概念、實現原理和應用
自動化在DevOps中的力量：簡化軟件開發和交付

自動化在DevOps中扮演著重要角色，它提升了DevOps的效能。通過自動化工具和方法，DevOps團隊可以實現以下目標：消除手動和重復性任務。簡化流程。在整個軟件開發生命周期中實現更
.NET 程序的 GDI 句柄泄露的再反思

一、背景1. 講故事上個月我寫過一篇如何洞察 C# 程序的 GDI 句柄泄露文章，當時用的是 GDIView + WinDbg 把問題搞定，前者用來定位泄露資源，后者用來定位泄露代碼，后面有朋友反
品牌洞察丨服務本地，美團直播成效幾何？

來源：17PR7月11日，美團App首頁推薦位出現“美團直播”的固定入口。在直播聚合頁面，外賣“神槍手”直播間、美團旅行直播間、美團買菜直播間等均已上線，同時
四年持續更迭堅持探索行業無人之境，HarmonyOS 4帶來五大升級多項創新

除了華為每年新發布的旗艦手機系列，上億花粉更加期待鴻蒙系統每次的跨版本大更新。8月4日，HarmonyOS 4于HDC 2023正式發布，這也是該系統歷經四年的再
Android 14發布：首批適配機型公布

5月11日消息，谷歌在今天凌晨舉行了I/O大會，本次發布會谷歌帶來了自家的AI語言模型PaLM 2、谷歌Pixel Fold折疊屏、谷歌Pixel 7a手機，同時發布了Androi
OPPO K11評測：旗艦級IMX890加持 2000元檔最強影像手機

【Techweb評測】中端機型用戶群體巨大，占了中國目前手機市場的大頭，一直以來都是各手機品牌的“必爭之地”，其中OPPO K系列機型一直以來都以高品質、

日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

SpringBatch高階應用：大數據批處理框架實戰指南

1. 環境準備

1.1 引入依賴

2.2 配置Job

2. 高階配置管理

2.1 通過Controller接口啟動Job

2.2 異步啟動Job

2.3 重啟Job

2.4 多線程執行Step

2.5 重復啟動Job

石頭自清潔掃拖機器人G10S評測：多年黑科技集大成之作懶人終極福音

0糖0卡0脂旭日森林仙草烏龍茶優惠：15瓶到手29元

三言兩語說透柯里化和反柯里化

自動化在DevOps中的力量：簡化軟件開發和交付

.NET 程序的 GDI 句柄泄露的再反思

品牌洞察丨服務本地，美團直播成效幾何？

四年持續更迭堅持探索行業無人之境，HarmonyOS 4帶來五大升級多項創新

Android 14發布：首批適配機型公布

OPPO K11評測：旗艦級IMX890加持 2000元檔最強影像手機

最新推薦

猜你喜歡

熱門推薦

相關資訊