多线程批处理数据导入工具 Java

傷城~ 2022-06-06 13:17 196阅读 0赞

由于在工作中经常需要进行数据的导入…..

调用

package com.lb.impl.qikandaohang.changecls;
    
    import java.sql.ResultSet;
    import java.sql.SQLException;
    
    import org.apache.commons.lang3.StringUtils;
    
    import com.lb.template.bantchexecute.QueryAndUpdateOneTemplateV2;
    import com.lb.util.Database;
    
    public class ChageImpactFactor extends QueryAndUpdateOneTemplateV2 { 
    
        public ChageImpactFactor(Database srcDb, Database descDb) {
            super(srcDb, descDb);
        }
    
        public static void main(String[] args) {
            Database srcDb = new Database("sa", "123456", "jdbc:sqlserver://192.168.2.44:1433;DatabaseName=Ixuehsu", com.lb.util.Constants.SQLSERVER_DRIVER);
            Database desc = new Database("sa", "123456", "jdbc:sqlserver://192.168.2.203:1433;DatabaseName=DB_wulixi", com.lb.util.Constants.SQLSERVER_DRIVER);
            String querySql = "SELECT ISSN, [封面], [综合影响因子] FROM [期刊封面]";
            QueryAndUpdateOneTemplateV2 template = new ChageImpactFactor(srcDb, desc);
            template.execute(querySql);
        }
    
        @Override
        public String getOprationSql(ResultSet resultSet) throws SQLException {
            String issn = resultSet.getString("ISSN").trim();
            String cover = getCover(resultSet.getString("封面"));
            String impactfactory = getImfactory(resultSet.getString("综合影响因子"));
    
            if (StringUtils.isNotBlank(issn)) {
                String updateSql = "UPDATE wlx_journal SET cover = '" + cover + "', impact_factory = '" + impactfactory + "' WHERE ISSN LIKE '%" + issn + "%';";
                return updateSql;
            }
    
            return null;
        }
    
        private String getImfactory(String string) {
            if (StringUtils.isNotBlank(string)) {
                return string;
            } else {
                return "0.0";
            }
        }
    
        private static String getCover(String string) {
            try {
                String url = string.substring(string.indexOf("/") + 1, string.indexOf("."));
                return url;
            } catch (Exception e) {
                return "";
            }
        }
    }

package com.lb.template.bantchexecute;
    
    import java.sql.Connection;
    import java.sql.PreparedStatement;
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.atomic.AtomicInteger;
    
    import org.apache.commons.lang3.StringUtils;
    
    import com.lb.util.DBManager;
    import com.lb.util.Database;
    
    public abstract class QueryAndUpdateOneTemplateV2 { 
    
        private DBManager srcDbMg;
        private DBManager descDbMg;
        private static AtomicInteger totalCount = new AtomicInteger();
    
        /** * 一次执行多少条 sql */
        private static final int SQL_BUFFER_COUNT = 1;
    
        /** * 最大线程数 */
        private static final int MAX_THREAD_COUNT = 1;
    
        private static final String END_FLAG = "_END_";
    
        private BlockingQueue<String> sqlsQueue = new ArrayBlockingQueue<>(10000);
    
        public QueryAndUpdateOneTemplateV2(Database srcDb, Database descDb) {
            this.srcDbMg = new DBManager(srcDb.getUserName(), srcDb.getPassword(), srcDb.getUrl(), srcDb.getDriver());
            this.descDbMg = new DBManager(descDb.getUserName(), descDb.getPassword(), descDb.getUrl(), descDb.getDriver());
        }
    
        public void execute(String querySql) {
            long start = System.currentTimeMillis();
            Reader reader = new Reader(querySql);
            reader.start();
    
            for (int i = 0; i < MAX_THREAD_COUNT; i++) {
                Writer writer = new Writer();
                writer.start();
            }
    
            long end = System.currentTimeMillis();
            // System.out.println("[修改完成], 共修改:" + totalCount.get() + ",耗时：" + (end - start) + ",毫秒");
        }
    
        /** * sql 读取者 * @author admin * */
        private class Reader extends Thread{ 
    
            private String querySql;
    
            public Reader(String querySql) {
                this.querySql = querySql;
            }
    
            @Override
            public void run() {
                System.out.println("读取线程启动....");
                Connection srcConn = srcDbMg.getConnection();
                int queryCount = 0;
    
                if (StringUtils.isBlank(querySql)) {
                    System.out.println("查询语句不为空");
                    return;
                }
    
                try {
                    PreparedStatement pstm = srcConn.prepareStatement(querySql);
                    ResultSet resultSet = pstm.executeQuery();
    
                    System.out.println("读取线程 开始读取...");
    
                    while (resultSet.next()) {
                        addOneSqlToBuffer(resultSet, sqlsQueue);
                        queryCount += 1;
                    }
    
                    sqlsQueue.put(END_FLAG);
                    System.out.println("读取完成, 共生成sql:" + queryCount + "次");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    
        /** * sql 写入者 * * @author admin * */
        private class Writer extends Thread{ 
    
            private List<String> sqlBuffer = new ArrayList<String>();
            private int executeCount;
    
            @Override
            public void run() {
                System.out.println("写入线程:" + Thread.currentThread().getName() + "启动...");
                int count = 0;
    
                while(true) {
                    try {
                        String sql = sqlsQueue.take();
    
                        if (StringUtils.isNotBlank(sql) && sql.equals(END_FLAG)) {
                            sqlsQueue.put(END_FLAG);
                            break;
                        }
    
                        sqlBuffer.add(sql);
                        count += 1;
    
                        if (count >= SQL_BUFFER_COUNT) {
                            long start1 = System.currentTimeMillis();
                            descDbMg.executes(sqlBuffer);
                            long end1 = System.currentTimeMillis();
                            executeCount += SQL_BUFFER_COUNT;
                            System.out.println(Thread.currentThread().getName() + "已修改:" + executeCount + ", 耗时：" + (end1 - start1));
    
                            totalCount.addAndGet(SQL_BUFFER_COUNT);
                            sqlBuffer.clear();
                            count = 0;
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        sqlBuffer.clear();
                        count = 0;
                        continue;
                    }
                }
    
                if (sqlBuffer.size() != 0) {
                    descDbMg.executes(sqlBuffer);
                    executeCount = executeCount + sqlBuffer.size();
                }
    
                System.out.println(Thread.currentThread().getName() + "[修改完成], 共修改:" + executeCount);
            }
        }
    
        private void addOneSqlToBuffer(ResultSet resultSet, BlockingQueue<String> sqlsQueue) throws SQLException {
            String sql = getOprationSql(resultSet);
    
            if (StringUtils.isNotBlank(sql)) {
                try {
                    sqlsQueue.put(sql);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    
        /** * 通过 查询结果集 生成 操作语句 * * @param resultSet * @return * @throws SQLException */
        public abstract String getOprationSql(ResultSet resultSet) throws SQLException;
    }

--------------------

package com.lb.util;
    
    import java.sql.Connection;
    import java.sql.DriverManager;
    import java.sql.PreparedStatement;
    import java.sql.ResultSet;
    import java.sql.SQLException;
    import java.sql.Statement;
    import java.util.List;
    import java.util.Map;
    
    public class DBManager { 
    
        private String userName;
        private String password;
        private String url;
        private String driver;
    
        public DBManager(String userName, String password, String url, String driver) {
            this.userName = userName;
            this.password = password;
            this.url = url;
            this.driver = driver;
        }
    
        public Connection getConnection() {
            Connection conn = null;
    
            try {
                Class.forName(driver);
                conn = DriverManager.getConnection(url, userName, password);
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            } catch (SQLException e) {
                e.printStackTrace();
            }
    
            return conn;
        }
    
        public void close(ResultSet rs, Statement statement, Connection conn) {
            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public void close(ResultSet rs, PreparedStatement ps, Connection conn) {
            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    
        /** * 批量执行多条 sqls, 一条失败全部回滚 */
        public void executes(List<String> sqls) {
            Connection conn = null;
            Statement statement = null;
            conn = getConnection();
            // 禁用自动提交
            try {
                statement = conn.createStatement();
                conn.setAutoCommit(false);
    
                for (String sql : sqls) {
                    statement.addBatch(sql);
                }
    
                statement.executeBatch();
    
                // 提交事务
                conn.commit();
                conn.setAutoCommit(true);
            } catch (SQLException e) {
                for (String sql : sqls) {
                    System.out.println(sql);
                }
                System.out.println("出错了");
    
                e.printStackTrace();
                if (conn != null) {
                    // 事务回滚
                    try {
                        conn.rollback();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                }
            } finally {
                close(null, statement, conn);
            }
        }
    
        /** * 执行多条 一个模板的多条sql语句, 一条失败全部回滚 (测试没通过) * * String : sql * Object[] : params */
        public void batchExecuteByTempalte(String sqlTemplate, List<Object[]> params) throws RuntimeException{
            Connection conn = null;
            PreparedStatement ps = null;
            conn = getConnection();
            // 禁用自动提交
            try {
                ps = conn.prepareStatement(sqlTemplate);
                conn.setAutoCommit(false);
    
                int paramSize = params.size();
                Object[] param = null;
                for (int i = 0; i < paramSize; i++) {
                    param = params.get(i);
                    for (int j = 0; j < param.length; j++) {
                        ps.setObject(j + 1, param[j]);
                    }
    
                    ps.addBatch();
                }
    
                ps.executeBatch();
                conn.commit();
                conn.setAutoCommit(true);
            } catch (SQLException e) {
                if (conn != null) {
                    // 事务回滚
                    try {
                        conn.rollback();
                        throw new RuntimeException();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                }
            } finally {
                close(null, ps, conn);
            }
        }
    
        /** * 执行多条 sqls, 一条失败全部回滚 * sql 不可重复 * String : sql * Object[] : params */
        public void executeWithParams(Map<String, Object[]> sqlMap) throws RuntimeException{
            Connection conn = null;
            PreparedStatement ps = null;
            conn = getConnection();
            // 禁用自动提交
            try {
                conn.setAutoCommit(false);
    
                for (Map.Entry<String, Object[]> entry : sqlMap.entrySet()) {
                    String sql = entry.getKey();
                    ps = conn.prepareStatement(sql);
                    Object[] params = entry.getValue();
    
                    for (int i = 0; i < params.length; i++) {
                        ps.setObject(i + 1, params[i]);
                    }
    
                    ps.execute();
                }
    
                // 提交事务
                conn.commit();
            } catch (SQLException e) {
                if (conn != null) {
                    // 事务回滚
                    try {
                        conn.rollback();
                        throw new RuntimeException();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                }
            } finally {
                close(null, ps, conn);
            }
        }
    
        /** * Map<Object[] : params * String: String * * @param sqlMaps * @throws RuntimeException */
        public void executeWithParamsNew(Map<Object[], String> sqlMap) throws RuntimeException{
            Connection conn = null;
            PreparedStatement ps = null;
            conn = getConnection();
            String errorSql = "";
            // 禁用自动提交
            try {
                conn.setAutoCommit(false);
    
                for (Map.Entry<Object[], String> entry : sqlMap.entrySet()) {
                    String sql =  entry.getValue();
                    errorSql = sql;
                    ps = conn.prepareStatement(sql);
                    Object[] params = entry.getKey();
    
                    for (int i = 0; i < params.length; i++) {
                        ps.setObject(i + 1, params[i]);
                    }
    
                    ps.execute();
                }
    
                // 提交事务
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
                System.out.println("error sql is:" + errorSql);
                if (conn != null) {
                    // 事务回滚
                    try {
                        conn.rollback();
                        throw new RuntimeException();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                }
            } finally {
                close(null, ps, conn);
            }
        }
    
        /** * 适合单表 * * Map<Object[] : params * String: String * * @param sqlMaps * @throws RuntimeException */
        public void bantchExecuteOneTable(Map<Object[], String> sqlMap) throws RuntimeException{
            Connection conn = null;
            PreparedStatement ps = null;
            conn = getConnection();
            // 禁用自动提交
            try {
                conn.setAutoCommit(false);
    
                for (Map.Entry<Object[], String> entry : sqlMap.entrySet()) {
                    String sql =  entry.getValue();
                    ps = conn.prepareStatement(sql);
                    Object[] params = entry.getKey();
    
                    for (int i = 0; i < params.length; i++) {
                        ps.setObject(i + 1, params[i]);
                    }
    
                    ps.addBatch();
                }
    
                ps.executeBatch();
                // 提交事务
                conn.commit();
                conn.setAutoCommit(true);
            } catch (SQLException e) {
                e.printStackTrace();
                if (conn != null) {
                    // 事务回滚
                    try {
                        conn.rollback();
                        throw new RuntimeException();
                    } catch (SQLException e1) {
                        e1.printStackTrace();
                    }
                }
            } finally {
                close(null, ps, conn);
            }
        }
    }

--------------------

package com.lb.util;
    
    /** * 数据库对象 * * @author admin * */
    public class Database { 
    
        private String userName;
    
        private String password;
    
        private String url;
    
        private String driver;
    
        public Database() {}
    
        public Database(String userName, String password, String url, String driver) {
            super();
            this.userName = userName;
            this.password = password;
            this.url = url;
            this.driver = driver;
        }
    
        public String getUserName() {
            return userName;
        }
    
        public void setUserName(String userName) {
            this.userName = userName;
        }
    
        public String getPassword() {
            return password;
        }
    
        public void setPassword(String password) {
            this.password = password;
        }
    
        public String getUrl() {
            return url;
        }
    
        public void setUrl(String url) {
            this.url = url;
        }
    
        public String getDriver() {
            return driver;
        }
    
        public void setDriver(String driver) {
            this.driver = driver;
        }
    }