package com.hdkj.lt.core.plugins.thread;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import org.apache.commons.compress.utils.Lists;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Function;

/**
 * @Description:
 * @Author: wenhao
 * @Create: 2023/10/30 18:03
 * @Version: 1.0
 */
@Getter
@Setter
@Component
@RequiredArgsConstructor
public class MultiThreadExecuteService {

    @Value("${thread-pool.config.coreSize}")
    private Integer threadPoolSize;

    private ThreadPoolTaskExecutor executor;

    @PostConstruct
    public void initThreadPool() {
        executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(threadPoolSize);
        executor.setMaxPoolSize(threadPoolSize + 1);
        executor.setQueueCapacity(500);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("task-");
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(60);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
    }

    /**
     * 此方法只适用于通过查询结果批量保存的场景，如：
     * insert into ... select ... from ... limit ...
     *
     * @param total    要插入的数据总数
     * @param args     查询的条件对象
     * @param function 要执行的mapper方法
     * @param <R>      返回结果
     */
    public <R> List<R> saveBatch(Long total, Object args, MultiThreadSaveBatchFunction<Object, R> function) {
        List<CompletableFuture<Integer>> futures = Lists.newArrayList();
        int size = (int) Math.ceil((double) total / threadPoolSize);
        for (int i = 0; i < threadPoolSize; i++) {
            final int page = i;
            futures.add((CompletableFuture<Integer>) CompletableFuture.supplyAsync(() -> {
                int current = page * size;
                return function.apply(current, size, args);
            }, executor));
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        List<R> result = Lists.newArrayList();
        for (CompletableFuture<Integer> future : futures) {
            try {
                result.add((R) future.get());
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        return result;
    }

    @FunctionalInterface
    public interface MultiThreadSaveBatchFunction<T, R> {

        R apply(int current, int size, T t);

    }

    public <T, R> List<R> executeBatch(List<T> entityList, Function<List<T>, R> function) {
        List<CompletableFuture<Integer>> futures = Lists.newArrayList();
        int length = entityList.size();
        int taskCount = length / threadPoolSize;
        for (int i = 0; i < threadPoolSize; i++) {
            final int page = i;
            futures.add((CompletableFuture<Integer>) CompletableFuture.supplyAsync(() -> {
                List<T> saveBatchList;
                if (page == threadPoolSize - 1) {
                    saveBatchList = entityList.subList(page * taskCount, length);
                } else {
                    saveBatchList = entityList.subList(page * taskCount, (page + 1) * taskCount);
                }
                return function.apply(saveBatchList);
            }, executor));
        }
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        List<R> result = Lists.newArrayList();
        for (CompletableFuture<Integer> future : futures) {
            try {
                result.add((R) future.get());
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        }
        return result;
    }

}
