package com.inspect.job.util;

import com.inspect.job.domain.task.PatrolTaskInfo;
import com.inspect.job.task.JobMainTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class ParallelProcessor {
    private final Logger log = LoggerFactory.getLogger(ParallelProcessor.class);

    private static final int MAX_GROUP_SIZE = 100;

    private final ExecutorService threadPool = new ThreadPoolExecutor(
            8, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000));

    // 模拟处理逻辑
    public void processGroup(List<PatrolTaskInfo> group) {
        System.out.println("Thread " + Thread.currentThread().getName() +
                " processing group of size " + group.size());
        // 模拟耗时处理
        try { Thread.sleep(200); } catch (InterruptedException ignored) {}
    }

    public void run(List<PatrolTaskInfo> allTasks) {
        // Step 1: 分组（相同 deviceCode 不拆组）
        Map<String, List<PatrolTaskInfo>> deviceMap = new HashMap<>();
        for (PatrolTaskInfo task : allTasks) {
            deviceMap.computeIfAbsent(task.getDeviceCode(), k -> new ArrayList<>()).add(task);
        }

        List<List<PatrolTaskInfo>> taskGroups = new ArrayList<>();
        List<PatrolTaskInfo> currentGroup = new ArrayList<>();

        for (List<PatrolTaskInfo> group : deviceMap.values()) {
            if (group.size() > MAX_GROUP_SIZE) {
                taskGroups.add(group); // 超大组直接入列
            } else {
                if (currentGroup.size() + group.size() > MAX_GROUP_SIZE) {
                    taskGroups.add(currentGroup);
                    currentGroup = new ArrayList<>();
                }
                currentGroup.addAll(group);
            }
        }
        if (!currentGroup.isEmpty()) {
            taskGroups.add(currentGroup);
        }

        // Step 2: 多线程并发处理
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (List<PatrolTaskInfo> group : taskGroups) {
            futures.add(CompletableFuture.runAsync(() -> processGroup(group), threadPool));
        }

        // 等待所有任务完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        threadPool.shutdown();
    }
}
