123

许鹏程

2024-05-12 5a7cfb9266c7e03ccee44811b0fba1d4aceae6fd

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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
package com.product.data.center.service;
 
import cn.hutool.core.util.NumberUtil;
import com.product.core.entity.DataTableEntity;
import com.product.core.entity.FieldSetEntity;
import com.product.core.exception.BaseException;
import com.product.core.service.support.AbstractBaseService;
import com.product.core.spring.context.SpringMVCContextHolder;
import com.product.data.center.config.ErrorCode;
import com.product.data.center.entity.BatchConfigEntity;
import com.product.data.center.entity.BatchExecuteEntity;
import com.product.data.center.entity.QueryDataConfigEntity;
import com.product.data.center.service.ide.IBatchService;
import com.product.datasource.dao.Dao;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
import java.util.Calendar;
 
 
@Service
public class BatchService extends AbstractBaseService implements IBatchService {
    @Autowired
    private JournalManagerService journalManagerService;
 
    /**
     * 批处理执行
     *
     * @param batchConEntity
     */
    public void executeBatch(BatchConfigEntity batchConEntity, QueryDataAfterProcessing queryDataAfterProcessing) throws BaseException {
        Dao dao = null;
        if (batchConEntity.getSourceDbe() == null) {
            throw new BaseException(ErrorCode.BATCH_EXECUTE_ERROR_DATASOURCE_EMPTY);
        }
        if (batchConEntity.getQueryEntity() == null) {
            throw new BaseException(ErrorCode.BATCH_EXECUTE_ERROR_DATASOURCE_EMPTY);
        }
        try {
            dao = batchConEntity.getSourceDbe().getDao();
            QueryDataConfigEntity queryDataConfig = batchConEntity.getQueryEntity();
            int totalPage;
            try {
                //获取查询数据总跳数SQL
                String queryCountSql = queryDataConfig.getQueryTotalCountSql();
                //查询总条数
                FieldSetEntity fs = dao.getOne(queryCountSql, queryDataConfig.getParams());
                int totalCount = fs.getInteger("total_count");
                //计算出总页数
                totalPage = (int) Math.ceil(NumberUtil.div(totalCount, batchConEntity.getBatchDisposeSize()));
            } catch (Exception e) {
                throw e;
            }
            DataTableEntity dt;
            int minId;
            int maxId = -1;
            Calendar startC;
            StringBuilder curFilter = new StringBuilder(128);
            String preFilter = batchConEntity.getQueryEntity().getQueryFilter();
            for (int i = 1; i <= totalPage; i++) {
                startC = Calendar.getInstance();
                //查询每页数据
                if (i > 1) {
                    curFilter.setLength(0);
                    curFilter.append(preFilter);
                    if (!StringUtils.isEmpty(preFilter)) {
                        curFilter.append(" and ");
                    }
                    curFilter.append(queryDataConfig.getAutoIncrementPrimaryField()).append(">").append(maxId);
                    batchConEntity.getQueryEntity().setQueryFilter(curFilter.toString());
                }
                minId = -1;
                maxId = -1;
                dt = dao.getList(queryDataConfig.getQuerySql(1, batchConEntity.getBatchDisposeSize()), queryDataConfig.getParams());
                try {
                    if (batchConEntity.getDataDispose() != null) {
                        //数据处理
                        dt = batchConEntity.getDataDispose().invokeMethod(dt);
                    }
                    //本批次最大ID
                    minId = dt.getInt(0, queryDataConfig.getAutoIncrementPrimaryField());
                    maxId = dt.getInt(dt.getRows() - 1, queryDataConfig.getAutoIncrementPrimaryField());
                    queryDataAfterProcessing.queryDataAfterProcessing(dt, minId, maxId, batchConEntity.getSourceDbe(), batchConEntity.getTargetDbe(), startC);
                } catch (Exception e) {
                    //调用错误处理
                    if (totalPage > 1) {
                        if (minId == -1 || maxId == -1) {
                            // 没有正确获取到最小最大的ID值，说明查询过程当中出现了问题，那么就不能继续执行，否则就无法通过错误日志进行重新处理（本质原因是无法确认对应的范围）
                            throw e;
                        }
                        batchConEntity.getBatchError().method(new BatchExecuteEntity(dt, maxId, minId, i, journalManagerService.getStackTrace(e)));
                    } else {
                        // 若是处理的数据只有1页，那么一旦报错就无需进行之后的数据处理（因为没有），直接进入整体报错的判定范围
                        throw e;
                    }
                }
            }
        } catch (BaseException e) {
            e.printStackTrace();
            SpringMVCContextHolder.getSystemLogger().error(e);
            throw e;
        } catch (Exception e) {
            e.printStackTrace();
            SpringMVCContextHolder.getSystemLogger().error(e);
            throw e;
        } finally {
            if (dao != null) {
                dao.closeConnection();
            }
        }
//
    }
}