Beauty of Refactoring: when multithreaded batch processing task lifts the beam - Universal scaffold

origin

Recently, I was responsible for data archiving , You have to learn from multithreading ; As a concurrent counter artifact - The thread pool should be well entertained ,I can I up , The following is a general development scaffold designed by the author for a batch task, including ：
Flow chart Introduction 、 Code demonstration 、 Some black Technology

The scaffold has the following advantages ：

1. General template , Any batch task scenario can inherit
2. Focus on business logic development of batch tasks , There is no need to consider batch granularity control , Thread control
3. Support multiple granularity ： Time segmentation , Integer shard 【 For example, export by table 】
4. Provide convenience for extension , such as execAfter Business monitoring or indicator collection can be carried out

---

Emerge

Flow chart Introduction

code

@Getter
public enum TaskTimeUnit {
    

    /** *  minute  */
    MINUTES("minutes"),
    HOURS("hours"),
    DAYS("days"),
    MONTHS("months"),
    YEARS("years");

    private final String unit;

    private final static Map<String, TaskTimeUnit> ALL = new HashMap<>();

    static {
    
        TaskTimeUnit[] values = TaskTimeUnit.values();
        for (TaskTimeUnit val : values) {
    
            ALL.put(val.getUnit(), val);
        }
    }

    TaskTimeUnit(String unit) {
    
        this.unit = unit;
    }


    public static TaskTimeUnit instanceOf(String unit) {
    
        return ALL.get(unit);
    }
}


public abstract class EsTaskCtlService<T> {
    
    protected T param;

    /** * apollo  Business key */
    @Resource
    protected ApolloKey apolloKey;

    /** *  Parameter checking  * * @param t */
    protected void init(T t) {
    
        Objects.requireNonNull(t, " invalid parameter ");
        this.param = t;
    }

    /** *  The class content actually executed  */
    protected abstract void exec();


    public void run(T t) {
    
        init(t);
        exec();
    }

    /** *  Handle a single integer task  */
    protected void dealWithTask(Integer num) {
     }

    /** *  Handle tasks for a certain period of time  * * @param start * @param end */
    protected void dealWithTask(LocalDateTime start, LocalDateTime end) throws IOException {
     }


    /** *  Execute tasks concurrently between integer ranges with actual business  *  Average resource consumption ：50%  Can be modified by <code>getTableConcurrentNum</code> Make custom adaptations  * * @param start * @param end * @param threadPoolExecutor */
    protected void assignAndExecuteTask(int start, int end, ExecutorService threadPoolExecutor) {
    
        int tableConcurrentNum = getConcurrentNum();
        log.info(" Number of tasks assigned to the current machine | Number of threads : {}", tableConcurrentNum);
        try {
    
            int tempEnd = 0;
            while (tempEnd < end) {
    
                tempEnd = start + tableConcurrentNum;
                if (tempEnd > end) {
    
                    tempEnd = end;
                }
                List<Integer> of = OrderTableId.of(start, tempEnd);
                CountDownLatch countDownLatch = new CountDownLatch(of.size());
                for (Integer finalI : of) {
    
                    threadPoolExecutor.execute(() -> {
    
                        dealWithTask(finalI);
                        countDownLatch.countDown();
                    });
                }
                countDownLatch.await();
                start = tempEnd + 1;
            }
        } catch (Exception e) {
    
            log.error(e.getMessage(), e);
        }
    }

    /** *  Cycle end condition  * * @return */
    protected boolean isRun() {
    
        return false;
    }


    protected void spiltTaskWithUnit(LocalDateTime startTime, LocalDateTime endTime, TaskTimeUnit jumpUnit, Integer jumpNum) throws IOException {
    
        LocalDateTime tempEnd = null;
        while (isRun()) {
    
            tempEnd = next(startTime, jumpUnit, jumpNum);
            if (startTime.plusNanos(1).isAfter(endTime)) {
    
                break;
            }
            dealWithTask(startTime, tempEnd);

            startTime = next(startTime, jumpUnit, jumpNum);
        }
        execAfter(startTime, endTime);
    }

    /** * hook after exec task, such as deal with release the system memory or resources or log .etc * * @param startTime * @param endTime */
    protected void execAfter(LocalDateTime startTime, LocalDateTime endTime) {
    
    }

    protected LocalDateTime next(LocalDateTime startTime, TaskTimeUnit jumpUnit, Integer jumpNum) {
    
        switch (jumpUnit) {
    
            case MINUTES:
                return startTime.plusMinutes(jumpNum);
            case HOURS:
                return startTime.plusHours(jumpNum);
            case DAYS:
                return startTime.plusDays(jumpNum);
            case MONTHS:
                return startTime.plusMonths(jumpNum);
            case YEARS:
                return startTime.plusYears(jumpNum);
            default:
                return startTime;
        }
    }

    /** *  Get the concurrency of each task  * * @return */
    protected static int getConcurrentNum() {
    
        int stand = 10;
        int count = Runtime.getRuntime().availableProcessors();
        int half = count >> 1;
        //  Not enough 1 And a half 
        if (count < stand) {
    
            return half;
        }
        //  More 1 Hemitropic 10 integer 
        return stand - (half % stand) + half;

    }

How to use ？ As a batch task , It is natural to control parameters , Task granularity segmentation ; Inherit show my code：

---

Fall scattered

Repeated wheel building process , There are always some whimsical reconstruction ideas , More or less , There is a design , No, wait for inspiration , In the mighty coding In the context of , Leave yourself a stage where you can stand alone .

The beauty of the refactoring , Always creating ...