java之CSV大批量數(shù)據(jù)入庫的實現(xiàn)

更新時間：2023年02月05日 11:11:01 作者：邱可愛

本文主要介紹了java之CSV大批量數(shù)據(jù)入庫的實現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

需求

讀200+的CSV/EXCEL文件，按文件名稱存到不同數(shù)據(jù)庫

前期準(zhǔn)備

環(huán)境

maven + jdk8 + mysql

代碼展示

pom文件

    <dependencies>
        <!--https://mvnrepository.com/artifact/com.opencsv/opencsv -->
        <dependency>            //讀取csv文件 
            <groupId>com.opencsv</groupId>
            <artifactId>opencsv</artifactId>
            <version>${opencsv.version}</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--mybatis-plus 持久層-->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>

        <!-- velocity 模板引擎, Mybatis Plus 代碼生成器需要 -->
        <dependency>
            <groupId>org.apache.velocity</groupId>
            <artifactId>velocity-engine-core</artifactId>
            <version>${velocity.version}</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

關(guān)鍵代碼及思路

多線程處理數(shù)據(jù)，否則8k萬數(shù)據(jù)太慢了

創(chuàng)建線程池，個數(shù)的大小一般取決于自己電腦配置，以及I/O還是CPU密集型。

I/O密集型：CPU*2 + 1
CPU密集型：CPU + 1
Java查看CPU數(shù)目

public static final Integer N_Thread = Runtime.getRuntime().availableProcessors() * 2 + 1;

 ExecutorService service = Executors.newFixedThreadPool(N_Thread );
 List<CompletableFuture<Void>> futureList = new ArrayList<>();

數(shù)據(jù)進(jìn)行分批處理，如果數(shù)據(jù)一次過大，可能導(dǎo)致sql拼接失敗或者程序連接超時的問題。

            List<AirStationHourEntity> records = new ArrayList<>(table.values());
            int basic = 0, total = records.size();

            do{
                int finalBasic = basic;
                String finalTableName = tableName;
                CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
                    List<AirStationHourEntity> subRecords = records.subList(finalBasic * 6000, Math.min((1 + finalBasic) * 6000, total));
                    airStationHourMapper.saveOrUpdateBatch(finalTableName, subRecords);
                }, service);
                // 加入線程集合，方便后續(xù)阻塞主線程，防止線程沒跑就提前結(jié)束
                futureList.add(completableFuture);
                basic++;
            }while (basic * 6000 < total);

獲取文件數(shù)據(jù)

		String url = dataConfig.getStation();
        List<String> stationFiles = FileUtils.findFiles(url);
        //遍歷文件
        for (String fileName : stationFiles) {
            try {
                // 讀取文件數(shù)據(jù)
                String path = url.concat(fileName);
                CSVReader reader = new CSVReader(new InputStreamReader(new FileInputStream(path), StandardCharsets.UTF_8));
                List<String[]> lines = reader.readAll();

根據(jù)文件名創(chuàng)建相應(yīng)數(shù)據(jù)庫

 tableName = fileName.substring(fileName.lastIndexOf("_") + 1, fileName.indexOf(".") - 2);
                tableName = TABLE_PREFIX.STATION_PREFIX.value.concat(tableName);
                if (airStationHourMapper.findTableByName(tableName) == 0) {
                    airStationHourMapper.createNewTable(tableName);
                }

注意：如果在主線程完成之后，沒有對主線程進(jìn)行阻塞，會導(dǎo)致線程池中的線程沒跑完就直接結(jié)束了，需要遍歷線程集合來阻塞主線程

 for (CompletableFuture<Void> future : futureList) {
     future.join();
 }

關(guān)鍵的SQL語句

saveOrUpdateBatch

這里選用Mysql提供的ON DUPLICATE KEY UPDATE來實現(xiàn)更新或者插入，如果primary key 或者 unique key不存在就插入，否則就更新。注意：primary key 和 unique key都存在的時候可能會導(dǎo)致數(shù)據(jù)的更新的異常，這里建議選其中一個最為鍵，否則容易死鎖！見方案選型

 <update id="saveOrUpdateBatch">
        insert into ${tableName}
        (
        <include refid="Base_Column_List"/>
        )
        values
        <foreach collection="records" item="record" separator=",">
            (
            #{record.dataId,jdbcType=VARCHAR},
            #{record.stationId,jdbcType=VARCHAR},
            #{record.dataTime,jdbcType=TIMESTAMP},
            #{record.aqi,jdbcType=INTEGER},
            #{record.priPol,jdbcType=VARCHAR},
            #{record.co,jdbcType=DOUBLE},
            #{record.co24,jdbcType=DOUBLE},
            #{record.no2,jdbcType=INTEGER},
            #{record.no224,jdbcType=INTEGER},
            #{record.so2,jdbcType=INTEGER},
            #{record.so224,jdbcType=INTEGER},
            #{record.o3,jdbcType=INTEGER},
            #{record.o324,jdbcType=INTEGER},
            #{record.o38,jdbcType=INTEGER},
            #{record.o3824,jdbcType=INTEGER},
            #{record.pm10,jdbcType=INTEGER},
            #{record.pm1024,jdbcType=INTEGER},
            #{record.pm25,jdbcType=INTEGER},
            #{record.pm2524,jdbcType=INTEGER}
            )
        </foreach>
        ON DUPLICATE KEY UPDATE
        AQI = VALUES(AQI),pri_pol = VALUES(pri_pol),
        CO = VALUES(CO),CO_24 = VALUES(CO_24),
        NO2 = VALUES(NO2),NO2_24 = VALUES(NO2_24),
        SO2 = VALUES(SO2),SO2_24 = VALUES(SO2_24),
        O3 = VALUES(O3),O3_8 = VALUES(O3_8),O3_8_24 = VALUES(O3_8_24),
        PM10 = VALUES(PM10),PM10_24 = VALUES(PM10_24),
        PM2_5 = VALUES(PM2_5),PM2_5_24 =VALUES(PM2_5_24);
    </update>

方案選型

此方案時間過久，10多個小時大概能完成100w+的數(shù)據(jù)；

在程序中查詢數(shù)據(jù)庫中現(xiàn)存的數(shù)據(jù)，然后對這些數(shù)據(jù)進(jìn)行更新，其余的進(jìn)行插入

此方案程序代碼看起來較繁瑣。

選用 ON DUPLICATE KEY UPDATE & 多線程來實現(xiàn)批量處理。問題在處理最后一些數(shù)據(jù)時，報異常：獲取不到數(shù)據(jù)庫連接，連接超時

解決方案

修改application.yml的sql配置

    hikari:
      connection-timeout: 600000  //時間設(shè)的長一些

當(dāng)即存在primary key 以及 unique key時，出現(xiàn)了死鎖

如果一個表定義有多個唯一鍵（包括唯一索引、主鍵）時，是不安全的。
當(dāng)mysql執(zhí)行INSERT ON DUPLICATE KEY的INSERT時，存儲引擎會檢查插入的行為是否產(chǎn)生重復(fù)錯誤。

org.springframework.dao.DeadlockLoserDataAccessException:
### Error updating database.  Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction
### The error may exist in class path resource [mapper/PersonGroupRefMapper.xml]
### The error may involve com.order.addOrder-Inline
### The error occurred while setting parameters
### SQL: insert into t_***(XX,XX,XX,XX,XX,XX) values         (?, ?, ?, ?, ?, ?)         ON DUPLICATE KEY UPDATE XX= VALUES(XX), XX= VALUES(XX), XX= VALUES(XX)
### Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transaction Deadlock found when trying to get lock; try restarting transaction; nested exception is com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Deadlock found when trying to get lock; try restarting transactio

解決方案

見上面關(guān)鍵SQL，刪除了primary key 留下了unique key重新建表。

批量數(shù)據(jù)入庫，當(dāng)SQL語句拼接過長，超過了設(shè)置的最大的限制。