您好,登錄后才能下訂單哦!
使用Spring Batch如何實(shí)現(xiàn)將txt文件寫(xiě)入數(shù)據(jù)庫(kù)?針對(duì)這個(gè)問(wèn)題,這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答,希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。
1、創(chuàng)建 Maven 項(xiàng)目,并在 pom.xml 中添加依賴
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.2.RELEASE</version> </parent> <properties> <java.version>1.8</java.version> </properties> <dependencies> <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> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.mybatis.spring.boot</groupId> <artifactId>mybatis-spring-boot-starter</artifactId> <version>1.2.0</version> </dependency> <!-- 工具類依賴--> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.12.6</version> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.4</version> </dependency> <!-- 數(shù)據(jù)庫(kù)相關(guān)依賴 --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <scope>runtime</scope> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid</artifactId> <version>1.0.26</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> </dependencies>
這里是這個(gè)小項(xiàng)目中用到的所有依賴,包括連接數(shù)據(jù)庫(kù)的依賴以及工具類等。
2、編寫(xiě) Model 類
我們要從文檔中讀取的有效列就是 uid,tag,type,就是用戶 ID,用戶可能包含的標(biāo)簽(用于推送),用戶類別(用戶用戶之間互相推薦)。
UserMap.java 中的 @Entity,@Column 注解,是為了利用 JPA 生成數(shù)據(jù)表而寫(xiě)的,可要可不要。
UserMap.java
@Data @EqualsAndHashCode @NoArgsConstructor @AllArgsConstructor //@Entity(name = "user_map") public class UserMap extends BaseModel { @Column(name = "uid", unique = true, nullable = false) private Long uid; @Column(name = "tag") private String tag; @Column(name = "type") private Integer type; }
3、實(shí)現(xiàn)批處理配置類
BatchConfiguration.java
@Configuration @EnableBatchProcessing public class BatchConfiguration { @Autowired public JobBuilderFactory jobBuilderFactory; @Autowired public StepBuilderFactory stepBuilderFactory; @Autowired @Qualifier("prodDataSource") DataSource prodDataSource; @Bean public FlatFileItemReader<UserMap> reader() { FlatFileItemReader<UserMap> reader = new FlatFileItemReader<>(); reader.setResource(new ClassPathResource("c152.txt")); reader.setLineMapper(new DefaultLineMapper<UserMap>() {{ setLineTokenizer(new DelimitedLineTokenizer("|") {{ setNames(new String[]{"uid", "tag", "type"}); }}); setFieldSetMapper(new BeanWrapperFieldSetMapper<UserMap>() {{ setTargetType(UserMap.class); }}); }}); return reader; } @Bean public JdbcBatchItemWriter<UserMap> importWriter() { JdbcBatchItemWriter<UserMap> writer = new JdbcBatchItemWriter<>(); writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()); writer.setSql("INSERT INTO user_map (uid,tag,type) VALUES (:uid, :tag,:type)"); writer.setDataSource(prodDataSource); return writer; } @Bean public JdbcBatchItemWriter<UserMap> updateWriter() { JdbcBatchItemWriter<UserMap> writer = new JdbcBatchItemWriter<>(); writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()); writer.setSql("UPDATE user_map SET type = (:type),tag = (:tag) WHERE uid = (:uid)"); writer.setDataSource(prodDataSource); return writer; } @Bean public UserMapItemProcessor processor(UserMapItemProcessor.ProcessStatus processStatus) { return new UserMapItemProcessor(processStatus); } @Bean public Job importUserJob(JobCompletionNotificationListener listener) { return jobBuilderFactory.get("importUserJob") .incrementer(new RunIdIncrementer()) .listener(listener) .flow(importStep()) .end() .build(); } @Bean public Step importStep() { return stepBuilderFactory.get("importStep") .<UserMap, UserMap>chunk(100) .reader(reader()) .processor(processor(IMPORT)) .writer(importWriter()) .build(); } @Bean public Job updateUserJob(JobCompletionNotificationListener listener) { return jobBuilderFactory.get("updateUserJob") .incrementer(new RunIdIncrementer()) .listener(listener) .flow(updateStep()) .end() .build(); } @Bean public Step updateStep() { return stepBuilderFactory.get("updateStep") .<UserMap, UserMap>chunk(100) .reader(reader()) .processor(processor(UPDATE)) .writer(updateWriter()) .build(); } }
prodDataSource 是假設(shè)用戶已經(jīng)設(shè)置好的,如果不知道怎么配置,也可以參考之前的文章進(jìn)行配置:Springboot 集成 Mybatis。
reader(),這方法從文件中讀取數(shù)據(jù),并且設(shè)置了一些必要的參數(shù)。緊接著是寫(xiě)操作 importWriter()
和 updateWriter()
,讀者看其中一個(gè)就好,因?yàn)槲疫@里是需要更新或者修改的,所以分為兩個(gè)。
processor(ProcessStatus status)
,該方法是對(duì)我們處理數(shù)據(jù)的類進(jìn)行實(shí)例化,這里我根據(jù) status 是 IMPORT 還是 UPDATE 來(lái)獲取不同的處理結(jié)果。
其他的看代碼就可以看懂了,哈哈,不詳細(xì)說(shuō)了。
4、將獲得的數(shù)據(jù)進(jìn)行清洗
UserMapItemProcessor.java
public class UserMapItemProcessor implements ItemProcessor<UserMap, UserMap> { private static final int MAX_TAG_LENGTH = 200; private ProcessStatus processStatus; public UserMapItemProcessor(ProcessStatus processStatus) { this.processStatus = processStatus; } @Autowired IUserMapService userMapService; private static final String TAG_PATTERN_STR = "^[a-zA-Z0-9\\u4E00-\\u9FA5_-]+$"; public static final Pattern TAG_PATTERN = Pattern.compile(TAG_PATTERN_STR); private static final Logger LOG = LoggerFactory.getLogger(UserMapItemProcessor.class); @Override public UserMap process(UserMap userMap) throws Exception { Long uid = userMap.getUid(); String tag = cleanTag(userMap.getTag()); Integer label = userMap.getType() == null ? Integer.valueOf(0) : userMap.getType(); if (StringUtils.isNotBlank(tag)) { Map<String, Object> params = new HashMap<>(); params.put("uid", uid); UserMap userMapFromDB = userMapService.selectOne(params); if (userMapFromDB == null) { if (this.processStatus == ProcessStatus.IMPORT) { return new UserMap(uid, tag, label); } } else { if (this.processStatus == ProcessStatus.UPDATE) { if (!tag.equals(userMapFromDB.getTag()) && !label.equals(userMapFromDB.getType())) { userMapFromDB.setType(label); userMapFromDB.setTag(tag); return userMapFromDB; } } } } return null; } /** * 清洗標(biāo)簽 * * @param tag * @return */ private static String cleanTag(String tag) { if (StringUtils.isNotBlank(tag)) { try { tag = tag.substring(tag.indexOf("{") + 1, tag.lastIndexOf("}")); String[] tagArray = tag.split(","); Optional<String> reduce = Arrays.stream(tagArray).parallel() .map(str -> str.split(":")[0]) .map(str -> str.replaceAll("\'", "")) .map(str -> str.replaceAll(" ", "")) .filter(str -> TAG_PATTERN.matcher(str).matches()) .reduce((x, y) -> x + "," + y); Function<String, String> str = (s -> s.length() > MAX_TAG_LENGTH ? s.substring(0, MAX_TAG_LENGTH) : s); return str.apply(reduce.get()); } catch (Exception e) { LOG.error(e.getMessage(), e); } } return null; } protected enum ProcessStatus { IMPORT, UPDATE; } public static void main(String[] args) { String distinctTag = cleanTag("Counter({'《重新定義》': 3, '輕想上的輕小說(shuō)': 3, '小說(shuō)': 2, 'Fate': 2, '同人小說(shuō)': 2, '雪狼八組': 1, " + "'社會(huì)': 1, '人文': 1, '短篇': 1, '重新定義': 1, 'AMV': 1, '《FBD》': 1, '《雪狼六組》': 1, '戰(zhàn)爭(zhēng)': 1, '《灰羽聯(lián)盟》': 1, " + "'誰(shuí)說(shuō)輕想沒(méi)人寫(xiě)小說(shuō)': 1})"); System.out.println(distinctTag); } }
讀取到的數(shù)據(jù)格式如 main()
方法所示,清理之后的結(jié)果如:
輕想上的輕小說(shuō),小說(shuō),Fate,同人小說(shuō),雪狼八組,社會(huì),人文,短篇,重新定義,AMV,戰(zhàn)爭(zhēng),誰(shuí)說(shuō)輕想沒(méi)人寫(xiě)小說(shuō) 。
去掉了特殊符號(hào)以及數(shù)字等。使用了 Java8 的 Lambda 表達(dá)式。
并且這里在處理的時(shí)候,判斷如果該數(shù)據(jù)用戶已經(jīng)存在,則進(jìn)行更新,如果不存在,則新增。
5、Job 執(zhí)行結(jié)束回調(diào)類
JobCompletionNotificationListener.java
@Component public class JobCompletionNotificationListener extends JobExecutionListenerSupport { private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class); private final JdbcTemplate jdbcTemplate; @Autowired public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) { this.jdbcTemplate = jdbcTemplate; } @Override public void afterJob(JobExecution jobExecution) { System.out.println("end ....."); } }
具體的邏輯可自行實(shí)現(xiàn)。
完成以上幾個(gè)步驟,運(yùn)行項(xiàng)目,就可以讀取并寫(xiě)入數(shù)據(jù)到數(shù)據(jù)庫(kù)了。
關(guān)于使用Spring Batch如何實(shí)現(xiàn)將txt文件寫(xiě)入數(shù)據(jù)庫(kù)問(wèn)題的解答就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,如果你還有很多疑惑沒(méi)有解開(kāi),可以關(guān)注億速云行業(yè)資訊頻道了解更多相關(guān)知識(shí)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。