把LangChain跑起來的三個方法
使用LangChain開發(fā)LLM應用時，需要機器進行GLM部署，好多同學第一步就被勸退了，那么如何繞過這個步驟先學習LLM模型的應用，對Langchain進行快速上手？本片講解3個把LangChain跑起來
本篇文章主要內(nèi)容：通過Spring Batch從一個庫中讀取數(shù)據(jù)進過處理后寫入到另外一個庫中。
<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>配置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) ; // 每次分頁查詢多少條數(shù)據(jù) builder.pageSize(10) ; builder.queryString("select u from User u where u.uid <= 50") ; builder.saveState(true) ; builder.name("userReader") ; return builder.build() ;}配置數(shù)據(jù)源，該數(shù)據(jù)源是用來寫入操作的
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用來寫入操作（當前庫的數(shù)據(jù)寫入到另外一個庫，上面的數(shù)據(jù)源）
@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處理器，數(shù)據(jù)從當前庫讀取處理后經(jīng)過處理后再寫入另外的庫中
@BeanItemProcessor<User, User> userProcessor() { return new ItemProcessor<User, User>() { @Override public User process(User item) throws Exception { System.out.printf("%s - 開始處理數(shù)據(jù)：%s%n", Thread.currentThread().getName(), item.toString()) ; // 模擬耗時操作 TimeUnit.SECONDS.sleep(1) ; // 在這里你可以對數(shù)據(jù)進行相應的處理。 return item ; } } ;}配置Step將ItemReader、ItemProcessor、ItemWriter串聯(lián)在一起。
@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實例的容器。它將邏輯上屬于一個流程的多個步驟組合在一起，并允許對所有步驟的全局屬性（如可重啟性）進行配置。作業(yè)配置包含：
@BeanJob userJob(Step userStep1, Step userStep2) { return jobs.get("userJob").start(userStep1).build();}以上是Spring Batch定義配置一個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。當你調(diào)用該接口時，你會發(fā)現(xiàn)接口一直不會返回，一直阻塞，下圖是Job的啟動序列
圖片
根據(jù)上圖能知道，當你調(diào)用run方法后，會等待整個Job退出狀態(tài)為FINISHED或者FAILED后才能結(jié)束。所以，你需要異步完成，以便 SimpleJobLauncher 立即返回給調(diào)用者。而正確的序列應該是如下：
圖片
上圖通過異步方式啟動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。
當一個Job正在執(zhí)行，由于斷電或者強制終止了程序。當程序恢復后你希望能夠接著程序終止前的進度繼續(xù)執(zhí)行，這時候你需要進行如下的操作（本人沒有發(fā)現(xiàn)有什么API能夠操作的，可能文檔沒看仔細）。
當程序非正常終止是，下面兩張表的狀態(tài)都是STARTED，END_TIME為null
batch_job_execution表
圖片
batch_step_execution表
圖片
想要重新啟動必須將上面的狀態(tài)修改為STOPPED，END_TIME字段設置上值（是什么值無所謂）。
然后我們就可以繼續(xù)使用上面的Controller接口啟動任務繼續(xù)執(zhí)行了。
為了加快程序的執(zhí)行，我們可以為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中使用的任何池化資源（如數(shù)據(jù)源）都可能對并發(fā)性設置限制。請確保這些資源池至少與步驟中所需的并發(fā)線程數(shù)一樣大。
通過上面配置線程池后，你將在控制臺看到如下輸出。
圖片
默認將有4個線程同時進行處理。可以通過如下配置進行調(diào)整
@BeanStep userStep1(ItemReader<User> userReader, ItemProcessor<User, User> userProcessor, ItemWriter<User> userWriter) { return steps.get("userStep1") // ... // 節(jié)流限制10，這里配置的大小應該與你的數(shù)據(jù)庫連接池大小及使用的線程池核心線程數(shù)一致。 .throttleLimit(10) .build() ;}要想重復啟動Job，我們可以在啟動Job時設置不同的JobParameters參數(shù)，只要參數(shù)不同那么就可以重復的啟動Job。如下示例：
@GetMapping("/start/{page}")public Object start(@PathVariable("page") Long page) throws Exception { Map<String, JobParameter> parameters = new HashMap<>() ; // 每次設置的參數(shù)值不同即可。 parameters.put("page", new JobParameter(page)) ; JobParameters jobParameters = new JobParameters(parameters) ; this.userJobLauncher.run(userJob, jobParameters) ; return "started" ;}以上是本篇文章的全部內(nèi)容，希望對你有幫助。
本文鏈接：http://www.www897cc.com/showinfo-26-87012-0.htmlSpringBatch高階應用：大數(shù)據(jù)批處理框架實戰(zhàn)指南
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： 架構(gòu)設計中如何應對接口級故障？
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請聯(lián)系：2376512515@qq.com 津ICP備18001702號
津公網(wǎng)安備 12010102000574號