jdbc操作 資料庫同步，全量，加入執行緒，批處理
阿新 • • 發佈：2019-01-29
動態資料庫的全量、增量同步，多執行緒增加效能，批處理。
程式碼可以直接執行：
同步的類：
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import 執行緒控制類：
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
public class InsertThread extends Thread {
private String sql_data;
private String sql_insert;
private Connection conB;
private Connection conA;
static int batchSize = 2500;
// 執行緒計數器
static private int threadCounts;
// 執行緒名稱池
static private String threadNames[];
static {
// 假設這裡允許系統同時執行最大執行緒數為10個
int maxThreadCounts = 5;
threadNames = new String[maxThreadCounts];
// 初始化執行緒名稱池
for (int i = 1; i <= maxThreadCounts; i++) {
threadNames[i - 1] = "子執行緒_" + i;
}
}
public InsertThread() {
// 臨界資源鎖定
synchronized (InsertThread.class) {
// 執行緒總數加1
threadCounts++;
// 從執行緒名稱池中取出一個未使用的執行緒名
for (int i = 0; i < threadNames.length; i++) {
if (threadNames[i] != null) {
String temp = threadNames[i];
// 名被佔用後清空
threadNames[i] = null;
// 初始化執行緒名稱
this.setName(temp);
break;
}
}
}
}
public void run() {
try {
Long start = System.currentTimeMillis();
Statement stmt_source = conA.createStatement();
ResultSet rs_sql_data = stmt_source.executeQuery(sql_data);
ResultSetMetaData rsmd = rs_sql_data.getMetaData();
PreparedStatement ps = conB.prepareStatement(sql_insert);
int columnCount=rsmd.getColumnCount();
int count=1;
while (rs_sql_data.next()) {
count++;
for(int k=1;k<=columnCount;k++){
ps.setString(k, rs_sql_data.getString(k));
}
ps.addBatch();
if(count % batchSize == 0) {
ps.executeBatch();
conB.commit();
}
}
ps.executeBatch();
conB.commit();
Long end = System.currentTimeMillis();
System.out.println(this.getName()+",耗時："+(end-start)/1000 + "秒");
stmt_source.close();
rs_sql_data.close();
ps.close();
} catch (Exception e) {
System.out.println(e);
} finally {
synchronized (InsertThread.class) {
// 釋放執行緒名稱
String[] threadName = this.getName().split("_");
// 執行緒名使用完後放入名稱池
threadNames[Integer.parseInt(threadName[1]) - 1] = this.getName();
// 執行緒執行完畢後減1
threadCounts--;
/*
* 通知其他被阻塞的執行緒，但如果其他執行緒要執行，則該同步塊一定要執行結束（即直
* 到釋放佔的鎖），其他執行緒才有機會執行，所以這裡的只是喚醒在此物件監視器上等待
* 的所有執行緒，讓他們從等待池中進入物件鎖池佇列中，而這些執行緒重新執行時它們一定
* 要先要得該鎖後才可能執行，這裡的notifyAll是不會釋放鎖的，試著把下面的睡眠語
* 句註釋去掉，即使你已呼叫了notify方法，發現CreateThread中的同步塊還是好
* 像一直處於物件等待狀態，其實呼叫notify方法後，CreateThread執行緒進入了物件鎖
* 池佇列中了，只要它一獲取到鎖，CreateThread所線上程就會真真的被喚醒並執行。
*/
InsertThread.class.notifyAll();
System.out.println("----" + this.getName() + " 所佔用資源釋放完畢，當前系統正在執行的子執行緒數："
+ threadCounts);
}
}
}
static public int getThreadCounts() {
synchronized (InsertThread.class) {
return threadCounts;
}
}
public InsertThread(String sql_data, String sql_insert, Connection conB, Connection conA) {
super();
this.sql_data = sql_data;
this.sql_insert = sql_insert;
this.conB = conB;
this.conA = conA;
// 臨界資源鎖定
synchronized (InsertThread.class) {
// 執行緒總數加1
threadCounts++;
// 從執行緒名稱池中取出一個未使用的執行緒名
for (int i = 0; i < threadNames.length; i++) {
if (threadNames[i] != null) {
String temp = threadNames[i];
// 名被佔用後清空
threadNames[i] = null;
// 初始化執行緒名稱
this.setName(temp);
break;
}
}
}
}
}