說(shuō)明

pro環(huán)境在程序啟動(dòng)時(shí)需要運(yùn)行50w update sql，在uat環(huán)境測(cè)試
1）分批1000每次，批量更新需要花時(shí)間120s左右；
2）采用下面類(lèi)似第二個(gè)測(cè)試用例(先把數(shù)據(jù)插入臨時(shí)表、臨時(shí)表同步主表、刪除臨時(shí)表)，最終只花了34s，其中插入花了14s，臨時(shí)表更新到主表花了18s多
3）繼承自SQLServerBulkCSVFileRecord進(jìn)行大容量復(fù)制，比CachedRowSetImpl方式更快；

代碼

package com.ydfind.driver.ok;

import com.microsoft.sqlserver.jdbc.SQLServerBulkCSVFileRecord;
import com.microsoft.sqlserver.jdbc.SQLServerBulkCopy;
import com.microsoft.sqlserver.jdbc.SQLServerBulkCopyOptions;
import com.microsoft.sqlserver.jdbc.SQLServerException;
import org.junit.Test;

import java.io.ByteArrayInputStream;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class MyBulkCopyTest {

    // 1w數(shù)據(jù)7332ms
    @Test
    public void testBulkInsert() throws SQLException {
        // drd notes:
        long time = System.currentTimeMillis();

        String connectionUrl = "jdbc:sqlserver://localhost:1433;databaseName=dev-data;user=sa;password=465628578";
        String tableName = "bulkCopyTest";
        String deleteSql = "if exists (select * from dbo.sysobjects where id = object_id(N'[dbo].["
                + tableName + "]') and OBJECTPROPERTY(id, N'IsUserTable') = 1) DROP TABLE [" + tableName + "]";
        String createSql = "create table " + tableName + " (c1 bigint, c2 varchar(20))";
        // 要插入的數(shù)據(jù)
        List<Object[]> list = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            Object[] objects = new Object[2];
            objects[0] = Long.valueOf(i);
            objects[1] = "name_" + i;
            list.add(objects);
        }
        MySQLServerBulkRecord bulkRecord = new MySQLServerBulkRecord(list);
        bulkRecord.addColumnMetadata(1, "c1", Types.BIGINT, 19, 0);
//        bulkRecord.addColumnMetadata(1, "c1", Types.BIGINT, 19, 0);
        bulkRecord.addColumnMetadata(2, "c2", Types.VARCHAR, 20, 0);

        try (Connection connection = DriverManager.getConnection(connectionUrl);
             Statement statement = connection.createStatement();){
            connection.setAutoCommit(false);
            // 卸了舊的表
            statement.execute(deleteSql);
            // 創(chuàng)建新表
            statement.execute(createSql);

            // BULK INSERT
            SQLServerBulkCopyOptions options = new SQLServerBulkCopyOptions();
            options.setTableLock(true);
//            options.setBatchSize(10_0000);
            SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(connection);
            bulkCopy.setBulkCopyOptions(options);
            bulkCopy.setDestinationTableName(tableName);

            bulkCopy.writeToServer(bulkRecord);

            connection.commit();
            bulkCopy.close();
        }
        System.out.println("cost time " + (System.currentTimeMillis() - time));
    }

    public class MySQLServerBulkRecord extends SQLServerBulkCSVFileRecord {
        private List<Object[]> dataList;
        private int cur;
        public MySQLServerBulkRecord(List<Object[]> dataList) throws SQLServerException {
            super(new ByteArrayInputStream(new byte[]{}), "UTF-8", ",", false);
            this.dataList = dataList;
            cur = -1;
        }

        @Override
        public boolean next() throws SQLServerException {
            return ++cur < dataList.size();
        }

        @Override
        public Object[] getRowData() throws SQLServerException {
            if (cur >= dataList.size()) {
                return null;
            }
            return dataList.get(cur);
        }

    }

    // 更新到bulkCopyTest：先插入到臨時(shí)表bulkCopyTestTemp，再u(mài)pdate到bulkCopyTest，再刪除臨時(shí)表
    // 1w數(shù)據(jù)cost time 4362
    @Test
    public void testBulkUpdate() throws SQLException {
        // drd notes:
        long time = System.currentTimeMillis();

        String connectionUrl = "jdbc:sqlserver://localhost:1433;databaseName=dev-data;user=sa;password=465628578";
        String tableName = "bulkCopyTestTemp";
        String tableNameTarget = "bulkCopyTest";
        String deleteSql = "if exists (select * from dbo.sysobjects where id = object_id(N'[dbo].["
                + tableName + "]') and OBJECTPROPERTY(id, N'IsUserTable') = 1) DROP TABLE [" + tableName + "]";
        String createSql = "create table " + tableName + " (c1 bigint, c2 varchar(20))";
        String updateSql = "update " + tableNameTarget + " with(TABLOCK)  set c2 = t2.c2 from " + tableName + " t2 with(nolock) where " + tableNameTarget + ".c1 = t2.c1";
        // 要插入的數(shù)據(jù)
        List<Object[]> list = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            Object[] objects = new Object[2];
            objects[0] = Long.valueOf(i);
            objects[1] = "name1_" + i;
            list.add(objects);
        }
        MySQLServerBulkRecord bulkRecord = new MySQLServerBulkRecord(list);
        bulkRecord.addColumnMetadata(1, "c1", Types.BIGINT, 19, 0);
//        bulkRecord.addColumnMetadata(1, "c1", Types.BIGINT, 19, 0);
        bulkRecord.addColumnMetadata(2, "c2", Types.VARCHAR, 20, 0);

        try (Connection connection = DriverManager.getConnection(connectionUrl);
             Statement statement = connection.createStatement();){
            connection.setAutoCommit(false);
            // 臨時(shí)表
            statement.execute(deleteSql);
            statement.execute(createSql);

            SQLServerBulkCopyOptions options = new SQLServerBulkCopyOptions();
            options.setTableLock(true);
            options.setBatchSize(10_0000);
            SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(connection);
            bulkCopy.setBulkCopyOptions(options);
            bulkCopy.setDestinationTableName(tableName);

            bulkCopy.writeToServer(bulkRecord);

            // 從臨時(shí)表更新到主表
            statement.execute(updateSql);
            // 刪除臨時(shí)表
            statement.execute(deleteSql);

            connection.commit();
            bulkCopy.close();
        }
        // 1w數(shù)據(jù)7332ms
        System.out.println("cost time " + (System.currentTimeMillis() - time));
    }
}

分析

第一個(gè)測(cè)試用例后，結(jié)果：

第二個(gè)測(cè)試用例結(jié)果

50w數(shù)據(jù)插入示例-本地電腦

package com.ydfind.driver;

import com.microsoft.sqlserver.jdbc.SQLServerBulkCopy;
import com.microsoft.sqlserver.jdbc.SQLServerBulkCopyOptions;
import com.ydfind.driver.ok.MySQLServerBulkRecord;
import org.junit.Test;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class BulkCopyInsertTest {
    private static String testTableName = "test_insert";
    private static String dropSql = "if exists (select * from dbo.sysobjects where id = object_id(N'[dbo].[" + testTableName + "]') and OBJECTPROPERTY(id, N'IsUserTable') = 1) DROP TABLE [" + testTableName + "]";
    private static String createTableName = "create table " + testTableName + "(c1 bigint, c2 varchar(20), c3 bigint)";

    @Test
    public void testInsert() throws SQLException {
        // 1.若存在表，則刪除表
        // 2.創(chuàng)建表
        // 3.插入10w數(shù)據(jù)
        // 4.記錄插入10w數(shù)據(jù)的時(shí)間輸出
        String url = "jdbc:sqlserver://localhost:1433;databaseName=rec-manager;user=sa;password=465628578";
        try (Connection conn = DriverManager.getConnection(url);
             Statement statement = conn.createStatement()) {
            statement.execute(dropSql);
            statement.execute(createTableName);

            // 插入10w條記錄
            long time = System.currentTimeMillis();
            int id = 0;
            String sql = "insert into " + testTableName + "(c1, c2, c3) values(%d,'%s',%d)";
            for (int i = 0; i < 500; i++) {
                for (int j = 0; j < 1000; j++) {
                    id++;
                    String sql1 = String.format(sql, id, "name1_" + id, id);
                    statement.addBatch(sql1);
                }
                statement.executeBatch();
            }
            System.out.println("batch insert cost time = " + (System.currentTimeMillis() - time));
        }
    }

    @Test
    public void testBulkCopyInsert() throws SQLException {
        // 1.若存在表，則刪除表
        // 2.創(chuàng)建表
        // 3.插入10w數(shù)據(jù)
        // 4.記錄插入10w數(shù)據(jù)的時(shí)間輸出
        String url = "jdbc:sqlserver://localhost:1433;databaseName=rec-manager;user=sa;password=465628578";
        try (Connection conn = DriverManager.getConnection(url);
             Statement statement = conn.createStatement()) {
            statement.execute(dropSql);
            statement.execute(createTableName);

            // 插入10w條記錄
            long time = System.currentTimeMillis();

            List<Object[]> list = new ArrayList<>();
            for (int i = 0; i < 50_0000; i++) {
                Object[] objects = new Object[3];
                objects[0] = Long.valueOf(i);
                objects[1] = "name1_" + i;
                objects[2] = Long.valueOf(i);
                list.add(objects);
            }
            MySQLServerBulkRecord bulkRecord = new MySQLServerBulkRecord(list);
            bulkRecord.addColumnMetadata(1, "c1", Types.BIGINT, 19, 0);
            bulkRecord.addColumnMetadata(2, "c2", Types.VARCHAR, 20, 0);
            bulkRecord.addColumnMetadata(3, "c3", Types.BIGINT, 19, 0);

            SQLServerBulkCopyOptions options = new SQLServerBulkCopyOptions();
            options.setTableLock(true);
            options.setBatchSize(10000);
            SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(conn);
            bulkCopy.setBulkCopyOptions(options);
            bulkCopy.setDestinationTableName(testTableName);

            bulkCopy.writeToServer(bulkRecord);

            bulkCopy.close();
            System.out.println("bulkCopy insert cost time = " + (System.currentTimeMillis() - time));
        }
    }
}

bulkCopy insert cost time = 25052
batch insert cost time = 222397

可見(jiàn)bulkCopy的方式可以提升接近8倍。把executeBatch()每次的數(shù)量從1000提高到1w，執(zhí)行時(shí)間也需要175495，接近3分鐘。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-632489.html

到了這里，關(guān)于SQLServerBulkCopy大容量插入、更新操作的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！