多线程如何实现事务回滚？一招帮你搞定！

我会带着你远行 2023-10-09 19:26 32阅读 0赞

#### 特别说明CountDownLatch ####

**CountDownLatch是一个类springboot自带的类，可以直接用**，变量AtomicBoolean 也是可以直接使用

#### CountDownLatch的用法 ####

CountDownLatch典型用法：

**1、某一线程在开始运行前等待n个线程执行完毕。** 将CountDownLatch的计数器初始化为`new CountDownLatch(n)`，每当一个任务线程执行完毕，就将计数器减1 `countdownLatch.countDown()`，当计数器的值变为0时，在CountDownLatch上`await()`的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。

**2、实现多个线程开始执行任务的最大并行性。** 注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的`CountDownLatch(1)`，将其计算器初始化为1，多个线程在开始执行任务前首先`countdownlatch.await()`，当主线程调用`countDown()`时，计数器变为0，多个线程同时被唤醒。

#### CountDownLatch(num) 简单说明 ####

new 一个 `CountDownLatch(num)` 对象

建立对象的时候 num 代表的是需要等待 num 个线程

// 建立对象的时候 num 代表的是需要等待 num 个线程
    //主线程
    CountDownLatch mainThreadLatch = new CountDownLatch(num);
    //子线程
    CountDownLatch rollBackLatch  = new CountDownLatch(1);

#### 主线程：mainThreadLatch.await() 和mainThreadLatch.countDown() ####

新建对象

CountDownLatch mainThreadLatch = new CountDownLatch(num);

卡住主线程，让其等待子线程，代码`mainThreadLatch.await()`，放在主线程里

mainThreadLatch.await();

![ef8ba6324a37b02941869548a83abfe5.png][]

代码`mainThreadLatch.countDown()`，放在子线程里，每一个子线程运行一到这个代码，意味着`CountDownLatch(num)`，里面的num-1（自动减一）

mainThreadLatch.countDown();

`CountDownLatch(num)`里面的num减到0，也就是`CountDownLatch(0)`，被卡住的主线程`mainThreadLatch.await()`，就会往下执行

![689fd2535f48124925e14c414e1461b6.png][]

#### 子线程：rollBackLatch.await() 和rollBackLatch.countDown() ####

新建对象，特别注意：子线程这个num就是1（关于只能为1的解答在后面）

CountDownLatch rollBackLatch  = new CountDownLatch(1);

卡住子线程，阻止每一个子线程的事务提交和回滚

rollBackLatch.await();

![8127b61186a432f1eabc7710d2dd4cd9.png][]

代码`rollBackLatch.countDown()；`放在主线程里，而且是放在主线程的等待代码`mainThreadLatch.await()；`后面。

rollBackLatch.countDown();

![dc2092d48246f91ee793c8a197f8ecaa.png][]

#### 为什么所有的子线程会在一瞬间就被所有都释放了？ ####

![2132d93a99eab363ee3417286650f708.png][]

#### 事务的回滚是怎么结合进去的？ ####

假设总共20个子线程，那么其中一个线程报错了怎么实现所有线程回滚。

引入变量

AtomicBoolean rollbackFlag = new AtomicBoolean(false)

和字面意思是一样的：根据 rollbackFlag 的true或者false 判断子线程里面，是否回滚。

首先我们确定的一点：**rollbackFlag 是所有的子线程都用着这一个判断**

![f8e3bdb9a1881ff738339661fbc2fcde.png][]

![3bc62663081d7ee0bcfe126fedd18346.png][]

#### 主线程类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;
    
    
    @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){
    //                System.out.println(thread.getState());
    //                thread.join();//是等待这个线程结束;
    //            }
    
                mainThreadLatch.await();
                //所有等待的子线程全部放开
                rollBackLatch.countDown();
    
                //是主线程等待子线程的终止。也就是说主线程的代码块中，如果碰到了t.join()方法，此时主线程需要等待（阻塞），等待子线程结束了(Waits for this thread to die.),才能继续执行t.join()之后的代码块。
    
    
                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;
        }
    }

#### 代码实际运用踩坑！！！！ ####

![ee5bb67c4f05673ceeee5f8dfca4cb75.png][]

还记得这里有个一批次处理多少数据么，我这边设置了20，实际到运用中的时候客户给了个20W的数据，我批次设置为20，那就有1W个子线程！！！！

这还不是最糟糕的，最糟糕的是每个子线程都会创建一个数据库连接，数据库直接被我搞炸了

![64832436e933c28183da0323c5916245.png][]

所以这里需要把：

int num = 20; //一个批次的数据有多少

改成：

int num = 20000; //一个批次的数据有多少

[ef8ba6324a37b02941869548a83abfe5.png]: https://img-blog.csdnimg.cn/img_convert/ef8ba6324a37b02941869548a83abfe5.png
[689fd2535f48124925e14c414e1461b6.png]: https://img-blog.csdnimg.cn/img_convert/689fd2535f48124925e14c414e1461b6.png
[8127b61186a432f1eabc7710d2dd4cd9.png]: https://img-blog.csdnimg.cn/img_convert/8127b61186a432f1eabc7710d2dd4cd9.png
[dc2092d48246f91ee793c8a197f8ecaa.png]: https://img-blog.csdnimg.cn/img_convert/dc2092d48246f91ee793c8a197f8ecaa.png
[2132d93a99eab363ee3417286650f708.png]: https://img-blog.csdnimg.cn/img_convert/2132d93a99eab363ee3417286650f708.png
[f8e3bdb9a1881ff738339661fbc2fcde.png]: https://img-blog.csdnimg.cn/img_convert/f8e3bdb9a1881ff738339661fbc2fcde.png
[3bc62663081d7ee0bcfe126fedd18346.png]: https://img-blog.csdnimg.cn/img_convert/3bc62663081d7ee0bcfe126fedd18346.png
[ee5bb67c4f05673ceeee5f8dfca4cb75.png]: https://img-blog.csdnimg.cn/img_convert/ee5bb67c4f05673ceeee5f8dfca4cb75.png
[64832436e933c28183da0323c5916245.png]: https://img-blog.csdnimg.cn/img_convert/64832436e933c28183da0323c5916245.png