jdbc批处理+事务处理 十万级数据导入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * 远程取数据的数据库
 * @author Administrator
 *
 */
public class DbJob {
    private String url = "jdbc:oracle:thin:@127.0.0.1:1521:ORCL";//远程数据库地址，这个我给换本地了，以防暴露了，哈哈
    private String username = "system";
    private String pw = "12345";
    public Connection getConnection() {
     try {
      Class.forName("oracle.jdbc.driver.OracleDriver");
      try {
       conn = DriverManager.getConnection(url, username, pw);
      } catch (SQLException e) {
       e.printStackTrace();
      }
     } catch (ClassNotFoundException e) {
      e.printStackTrace();
     }
     return conn;
   }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
/**
 * 目标数据库（需要插入的数据库）
 * @author Administrator
 *
 */
public class DbTarget {
    private String url = "jdbc:jtds:sqlserver://127.0.0.1:1433/bigdata";//这个地址也改了，也可以是远程数据库，注意，目标数据库的表需要先建立起来，因为两个数据库一个orcal，一个sqlserver，数据结构不同所以需要手动修改再建立表。sql我就不写了。
    private String username = "admin";
    private String pw = "54321";
    private Connection conn = null;
    public Connection getConnection() {
     try {
      Class.forName("net.sourceforge.jtds.jdbc.Driver");
      try {
       conn = DriverManager.getConnection(url, username, pw);
      } catch (SQLException e) {
       e.printStackTrace();
      }
     } catch (ClassNotFoundException e) {
      e.printStackTrace();
     }
     return conn;
   }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
/**
 * 数据导入
 * @author Administrator
 *
 */
public class ConvertDataFromData {
    DbJob dbJob;
    DbTarget dbTarget;

    public ConvertDataFromData() {
        dbTarget = new DbTarget();
        dbJob = new DbJob();
    }
    //导入数据
    public void go() {
        try  {
            Connection targetConn = dbTarget.getConnection();
            Connection jobConn = dbJob.getConnection();
            // 从job库移到target库
            //先把事务自动提交关闭
            targetConn.setAutoCommit(false);
            PreparedStatement jps = jobConn.prepareStatement("select * from region");
            ResultSet rs = jps.executeQuery();
            if (!rs.next()) return;
            PreparedStatement cps = targetConn.prepareStatement("insert into region(REGION_ID,REGION_FIRST_CODE) values (?,?)");
            int index = 0;
            long start = System.currentTimeMillis();
            do {
                cps.setInt(1, rs.getInt(1));
                cps.setString(2, doSome(rs.getString(2)));// 需要做处理的数据               
                cps.addBatch();
                if (index++ % 500 == 0) {
                //500条数据批处理
                    cps.executeBatch();
                }
            } while (rs.next());
            if (index % 500 != 0) {
                cps.executeBatch();
            }
            //提交事务
            targetConn.commit();
            rs.close();
            //简写了 关闭连接
            jobConn.close();
            targetConn.close();
            System.out.println("total of data is " + index + ", used " + ((System.currentTimeMillis() - start)/1000) +"  s");
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public String doSome(String str) {
        // TODO 逻辑处理
        return str;
    }

    public static void main(String[] args) {
        new ConvertDataFromData().go();
    }
}