多线程实现事务回滚
- 特别说明CountDownLatch
- CountDownLatch的用法
- CountDownLatch(num) 简单说明
- 主线程：() 和()
- 子线程：() 和()
- **为什么所有的子线程会在一瞬间就被所有都释放了？**
- 事务的回滚是怎么结合进去的？
- 假设总共20个子线程，那么其中一个线程报错了怎么实现所有线程回滚。
- 引入变量AtomicBoolean rollbackFlag = new AtomicBoolean(false)
- 主线程类Entry
- 子线程类WorkThread
- 代码实际运用踩坑！！！！
特别说明CountDownLatch
CountDownLatch是一个类springboot自带的类，可以直接用！！！！！，
变量AtomicBoolean 也是可以直接使用
CountDownLatch的用法
CountDownLatch典型用法：1、某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为new CountDownLatch(n)，每当一个任务线程执行完毕，就将计数器减1 ()，当计数器的值变为0时，在CountDownLatch上await()的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。
CountDownLatch典型用法：2、实现多个线程开始执行任务的最大并行性。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计算器初始化为1，多个线程在开始执行任务前首先()，当主线程调用countDown()时，计数器变为0，多个线程同时被唤醒。
CountDownLatch(num) 简单说明
// new 一个 CountDownLatch(num) 对象
// 建立对象的时候 num 代表的是需要等待 num 个线程
// 建立对象的时候 num 代表的是需要等待 num 个线程
//主线程
CountDownLatch mainThreadLatch = new CountDownLatch(num);
//子线程
CountDownLatch rollBackLatch = new CountDownLatch(1);
主线程：() 和()
//新建对象
CountDownLatch mainThreadLatch = new CountDownLatch(num);
//卡住主线程，让其等待子线程,代码(),放在主线程里
mainThreadLatch.await();
//代码()，放在子线程里，每一个子线程运行一到这个代码，意味着CountDownLatch(num)，里面的num-1（自动减一）
mainThreadLatch.countDown();
//CountDownLatch(num)里面的num减到0，也就是CountDownLatch(0)，被卡住的主线程()，就会往下执行
子线程：() 和()
//新建对象，特别注意：子线程这个num就是1(关于只能为1的解答在后面)
CountDownLatch rollBackLatch = new CountDownLatch(1);
//卡住子线程，阻止每一个子线程的事务提交和回滚
rollBackLatch.await();
//代码()；放在主线程里，而且是放在主线程的等待代码()；后面。
rollBackLatch.countDown();
为什么所有的子线程会在一瞬间就被所有都释放了？
事务的回滚是怎么结合进去的？
假设总共20个子线程，那么其中一个线程报错了怎么实现所有线程回滚。
引入变量AtomicBoolean rollbackFlag = new AtomicBoolean(false)
和字面意思是一样的：根据 rollbackFlag 的true或者false 判断子线程里面，是否回滚。
首先我们确定的一点：rollbackFlag 是所有的子线程都用着这一个判断
主线程类Entry
package org.apache.dolphinscheduler.api.utils;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.dolphinscheduler.api.controller.WorkThread;
import org.apache.dolphinscheduler.common.enums.DbType;
import org.springframework.web.bind.annotation.*;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
/*2021-06-10 wx*/
@RestController
@RequestMapping("importDatabase")
public class Entry {
/**
* @param dbid 数据库的id
* @param tablename 表名
* @param sftpFileName 文件名称
* @param head 是否有头文件
* @param splitSign 分隔符
* @param type 数据库类型
*/
private static String SFTP_HOST = "192.168.1.92";
private static int SFTP_PORT = 22;
private static String SFTP_USERNAME = "root";
private static String SFTP_PASSWORD = "rootroot";
private static String SFTP_BASEPATH = "/opt/testSFTP/";
@PostMapping("/thread")
@ResponseBody
public static JSONObject importDatabase(@RequestParam("dbid") int dbid
,@RequestParam("tablename") String tablename
,@RequestParam("sftpFileName") String sftpFileName
,@RequestParam("head") String head
,@RequestParam("splitSign") String splitSign
,@RequestParam("type") DbType type
,@RequestParam("heads") String heads
,@RequestParam("scolumns") String scolumns
,@RequestParam("tcolumns") String tcolumns ) throws Exception {
JSONObject obForRetrun = new JSONObject();
try {
JSONArray jsonArray = JSONArray.parseArray(tcolumns);
JSONArray scolumnArray = JSONArray.parseArray(scolumns);
JSONArray headsArray = JSONArray.parseArray(heads);
List<Integer> listInteger = getRrightDataNum(headsArray,scolumnArray);
JSONArray bodys = SFTPUtils.getSftpContent(SFTP_HOST,SFTP_PORT,SFTP_USERNAME,SFTP_PASSWORD,SFTP_BASEPATH,sftpFileName,head,splitSign);
int total = bodys.size();
int num = 20; //一个批次的数据有多少
int count = total/num;//周期
int lastNum =total- count*num;//余数
List<Thread> list = new ArrayList<Thread>();
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss:SS");
TimeZone t = sdf.getTimeZone();
t.setRawOffset(0);
sdf.setTimeZone(t);
Long startTime=System.currentTimeMillis();
int countForCountDownLatch = 0;
if(lastNum==0){//整除
countForCountDownLatch= count;
}else{
countForCountDownLatch= count + 1;
}
//子线程
CountDownLatch rollBackLatch = new CountDownLatch(1);
//主线程
CountDownLatch mainThreadLatch = new CountDownLatch(countForCountDownLatch);
AtomicBoolean rollbackFlag = new AtomicBoolean(false);
StringBuffer message = new StringBuffer();
message.append("报错信息：");
//子线程
for(int i=0;i<count;i++) {//这里的count代表有几个线程
Thread g = new Thread(new WorkThread(i,num,tablename,jsonArray,dbid,type,bodys,listInteger,mainThreadLatch,rollBackLatch,rollbackFlag,message ));
g.start();
list.add(g);
}
if(lastNum!=0){//有小数的情况下
Thread g = new Thread(new WorkThread(0,lastNum,tablename,jsonArray,dbid,type,bodys,listInteger,mainThreadLatch,rollBackLatch,rollbackFlag,message ));
g.start();
list.add(g);
}
// for(Thread thread:list){
// (());
// ();//是等待这个线程结束;
// }
mainThreadLatch.await();
//所有等待的子线程全部放开
rollBackLatch.countDown();
//是主线程等待子线程的终止。也就是说主线程的代码块中，如果碰到了()方法，此时主线程需要等待（阻塞），等待子线程结束了(Waits for this thread to die.),才能继续执行()之后的代码块。
Long endTime=System.currentTimeMillis();
System.out.println("总共用时: "+sdf.format(new Date(endTime-startTime)));
if(rollbackFlag.get()){
obForRetrun.put("code",500);
obForRetrun.put("msg",message);
}else{
obForRetrun.put("code",200);
obForRetrun.put("msg","提交成功！");
}
obForRetrun.put("data",null);
}catch (InterruptedException e){
e.printStackTrace();
obForRetrun.put("code",500);
obForRetrun.put("msg",e.getMessage());
obForRetrun.put("data",null);
}
return obForRetrun;
}
/**
* 文件里第几列被作为导出列
* @param headsArray
* @param scolumnArray
* @return
*/
public static List<Integer> getRrightDataNum(JSONArray headsArray, JSONArray scolumnArray){
List<Integer> list = new ArrayList<Integer>();
String arrayA [] = new String[headsArray.size()];
for(int i=0;i<headsArray.size();i++){
JSONObject ob = (JSONObject)headsArray.get(i);
arrayA[i] =String.valueOf(ob.get("title"));
}
String arrayB [] = new String[scolumnArray.size()];
for(int i=0;i<scolumnArray.size();i++){
JSONObject ob = (JSONObject)scolumnArray.get(i);
arrayB[i] =String.valueOf(ob.get("columnName"));
}
for(int i =0;i<arrayA.length;i++){
for(int j=0;j<arrayB.length;j++){
if(arrayA[i].equals(arrayB[j])){
list.add(i);
break;
}
}
}
return list;
}
}
子线程类WorkThread
package org.apache.dolphinscheduler.api.controller;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.dolphinscheduler.api.service.DataSourceService;
import org.apache.dolphinscheduler.common.enums.DbType;
import org.apache.dolphinscheduler.dao.entity.DataSource;
import org.apache.dolphinscheduler.dao.mapper.DataSourceMapper;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.springframework.transaction.PlatformTransactionManager;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 多线程
*/
public class WorkThread implements Runnable{ //建立线程的两种方法 1 实现Runnable 接口 2 继承 Thread 类
private DataSourceService dataSourceService;
private DataSourceMapper dataSourceMapper;
private Integer begin;
private Integer end;
private String tableName;
private JSONArray columnArray;
private Integer dbid;
private DbType type;
private JSONArray bodys;
private List<Integer> listInteger;
private PlatformTransactionManager transactionManager;
private CountDownLatch mainThreadLatch;
private CountDownLatch rollBackLatch;
private AtomicBoolean rollbackFlag;
private StringBuffer message;
/**
* @param i
* @param num
* @param tableFrom
* @param columnArrayFrom
* @param dbidFrom
* @param typeFrom
*/
public WorkThread(int i, int num, String tableFrom, JSONArray columnArrayFrom, int dbidFrom
, DbType typeFrom, JSONArray bodysFrom, List<Integer> listIntegerFrom
,CountDownLatch mainThreadLatch,CountDownLatch rollBackLatch,AtomicBoolean rollbackFlag
,StringBuffer messageFrom) {
begin=i*num;
end=begin+num;
tableName = tableFrom;
columnArray = columnArrayFrom;
dbid = dbidFrom;
type = typeFrom;
bodys = bodysFrom;
listInteger = listIntegerFrom;
this.dataSourceMapper = SpringApplicationContext.getBean(DataSourceMapper.class);
this.dataSourceService = SpringApplicationContext.getBean(DataSourceService.class);
this.transactionManager = SpringApplicationContext.getBean(PlatformTransactionManager.class);
this.mainThreadLatch = mainThreadLatch;
this.rollBackLatch = rollBackLatch;
this.rollbackFlag = rollbackFlag;
this.message = messageFrom;
}
public void run() {
DataSource dataSource = dataSourceMapper.queryDataSourceByID(dbid);
String cp = dataSource.getConnectionParams();
Connection con=null;
con = dataSourceService.getConnection(type,cp);
if(con!=null)
{
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss:SS");
TimeZone t = sdf.getTimeZone();
t.setRawOffset(0);
sdf.setTimeZone(t);
Long startTime = System.currentTimeMillis();
try {
con.setAutoCommit(false);
//---------------------------- 获取字段和类型
String columnString = null;//活动的字段
int intForType = 0;
String type[] = new String[columnArray.size()];//类型集合
for(int i=0;i<columnArray.size();i++){
JSONObject ob = (JSONObject)columnArray.get(i);
if(columnString==null){
columnString = String.valueOf(ob.get("name"));
}else{
columnString = columnString + ","+ String.valueOf(ob.get("name"));
}
type[intForType] = String.valueOf(ob.get("type"));
intForType = intForType + 1;
}
intForType = 0;
//这一步是为了形成 insert into "+tableName+"(id,name，age) values (?,?,?);
String dataString = null;
for(int i=0;i<columnArray.size();i++){
if(dataString==null){
dataString = "?";
}else{
dataString = dataString +","+"?";
}
}
//--------------------------------
StringBuffer sql = new StringBuffer();
sql = sql.append("insert into "+tableName+"("+columnString+") values ("+dataString+")") ;
PreparedStatement pst= (PreparedStatement)con.prepareStatement(sql.toString());
for(int i=begin;i<end;i++) {
JSONObject ob = (JSONObject)bodys.get(i);
if(ob!=null){
String [] array = ob.get(i).toString().split("\\,");
String [] arrayFinal = getFinalData(listInteger,array);
for(int j=0;j<type.length;j++){
String typeString = type[j].toLowerCase();
int z = j+1;
if("string".equals(typeString)||"varchar".equals(typeString)){
pst.setString(z,arrayFinal[j]);//这里的第一个参数 是指 替换第几个？
}else if("int".equals(typeString)||"bigint".equals(typeString)){
pst.setInt(z,Integer.valueOf(arrayFinal[j]));//这里的第一个参数 是指 替换第几个？
}else if("long".equals(typeString)){
pst.setLong(z,Long.valueOf(arrayFinal[j]));//这里的第一个参数 是指 替换第几个？
}else if("double".equals(typeString)){
pst.setDouble(z,Double.parseDouble(arrayFinal[j]));
}else if("date".equals(typeString)||"datetime".equals(typeString)){
pst.setDate(z, setDateback(arrayFinal[j]));
}else if("Timestamp".equals(typeString)){
pst.setTimestamp(z, setTimestampback(arrayFinal[j]));
}
}
}
pst.addBatch();
}
pst.executeBatch();
mainThreadLatch.countDown();
rollBackLatch.await();
if(rollbackFlag.get()){
con.rollback();//表示回滚事务；
}else{
con.commit();//事务提交
}
con.close();
} catch (Exception e) {
System.out.println(e.getMessage());
message = message.append(e.getMessage());
rollbackFlag.set(true);
mainThreadLatch.countDown();
try {
con.close();
} catch (SQLException throwables) {
throwables.printStackTrace();
}
}
Long endTime = System.currentTimeMillis();
System.out.println(Thread.currentThread().getName()+":startTime= "+sdf.format(new Date(startTime))+",endTime= "+sdf.format(new Date(endTime))
+" 用时："+sdf.format(new Date(endTime - startTime)));
}
}
public java.sql.Date setDateback(String dateString) throws ParseException {
SimpleDateFormat sdf = new SimpleDateFormat( "yyyy-MM-dd HH:mm:ss" );
java.util.Date date = sdf.parse( "2015-5-6 10:30:00" );
long lg = date.getTime();// 日期 转 时间戳
return new java.sql.Date( lg );
}
public java.sql.Timestamp setTimestampback(String dateString) throws ParseException {
SimpleDateFormat sdf = new SimpleDateFormat( "yyyy-MM-dd HH:mm:ss" );
java.util.Date date = sdf.parse( "2015-5-6 10:30:00" );
long lg = date.getTime();// 日期 转 时间戳
return new java.sql.Timestamp( lg );
}
public String [] getFinalData(List<Integer> listInteger,String[] array){
String [] arrayFinal = new String [listInteger.size()];
for(int i=0;i<listInteger.size();i++){
int a = listInteger.get(i);
arrayFinal[i] = array[a];
}
return arrayFinal;
}
}
代码实际运用踩坑！！！！
还记得这里有个一批次处理多少数据么，我这边设置了20，实际到运用中的时候客户给了个20W的数据，我批次设置为20，那就有1W个子线程！！！！
这还不是最糟糕的，最糟糕的是每个子线程都会创建一个数据库连接，数据库直接被我搞炸了
所以这里需要把：int num = 20; //一个批次的数据有多少
改成：
int num = 20000; //一个批次的数据有多少