import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import org.springframework.stereotype.Component;
 
import javax.annotation.PostConstruct;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
 
/**
 * 初始化隊(duì)列及線程池
 * @author tarzan
 *
 */
@Component
public class TaskQueue {
    //處理器隊(duì)列
    public static BlockingQueue<TaskProcessor> taskProcessors;
    //等待任務(wù)隊(duì)列
    public static BlockingQueue<CompileTask> waitTasks;
    //處理任務(wù)隊(duì)列
    public static BlockingQueue<CompileTask> executeTasks;
    //線程池
    public static ExecutorService exec;
    //初始處理器數(shù)(計(jì)算機(jī)cpu可用線程數(shù))
    public static Integer processorNum=Runtime.getRuntime().availableProcessors();
 
    /**
     * 初始化處理器、等待任務(wù)、處理任務(wù)隊(duì)列及線程池
     */
    @PostConstruct
    public static void initEquipmentAndUsersQueue(){
        exec = Executors.newCachedThreadPool();
        taskProcessors =new LinkedBlockingQueue<TaskProcessor>(processorNum);
        //將空閑的設(shè)備放入設(shè)備隊(duì)列中
        setFreeDevices(processorNum);
        waitTasks =new LinkedBlockingQueue<CompileTask>();
        executeTasks=new LinkedBlockingQueue<CompileTask>(processorNum);
    }
 
 
    /**
     * 將空閑的處理器放入處理器隊(duì)列中
     */
    private static void setFreeDevices(int num) {
        //獲取可用的設(shè)備
        for (int i = 0; i < num; i++) {
            TaskProcessor dc=new TaskProcessor();
            try {
                taskProcessors.put(dc);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
 
 
 
    public static CompileTask getWaitTask(Long clazzId) {
        return get(TaskQueue.waitTasks,clazzId);
    }
 
    public static CompileTask getExecuteTask(Long clazzId) {
        return get(TaskQueue.executeTasks,clazzId);
    }
 
 
    private static CompileTask get(BlockingQueue<CompileTask> users, Long clazzId) {
        CompileTask compileTask =null;
        if (CollectionUtils.isNotEmpty(users)){
            Optional<CompileTask> optional=users.stream().filter(e->e.getClazzId().longValue()==clazzId.longValue()).findFirst();
            if(optional.isPresent()){
                compileTask =  optional.get();
            }
        }
        return compileTask;
    }
 
    public static Integer getSort(Long clazzId) {
        AtomicInteger index = new AtomicInteger(-1);
        BlockingQueue<CompileTask> compileTasks = TaskQueue.waitTasks;
        if (CollectionUtils.isNotEmpty(compileTasks)){
            compileTasks.stream()
                    .filter(e -> {
                        index.getAndIncrement();
                        return e.getClazzId().longValue() == clazzId.longValue();
                    })
                    .findFirst();
        }
        return index.get();
    }
 
    //單位秒
    public static int estimatedTime(Long clazzId){
        return  estimatedTime(60,getSort(clazzId)+1);
    }
 
    //單位秒
    public static int estimatedTime(int cellMs,int num){
         int a= (num-1)/processorNum;
         int b= cellMs*(a+1);
        return  b;
    }
 
 
 
 
 

import lombok.Data;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.gis.common.enums.DataScheduleEnum;
import org.springblade.gis.dynamicds.service.DynamicDataSourceService;
import org.springblade.gis.modules.feature.schedule.service.DataScheduleService;
 
import java.util.Date;
 
 
@Data
public class CompileTask implements Runnable {
    //當(dāng)前請(qǐng)求的線程對(duì)象
    private Long clazzId;
    //用戶id
    private Long userId;
    //當(dāng)前請(qǐng)求的線程對(duì)象
    private Thread thread;
    //綁定處理器
    private TaskProcessor taskProcessor;
    //任務(wù)狀態(tài)
    private Integer status;
    //開始時(shí)間
    private Date startTime;
    //結(jié)束時(shí)間
    private Date endTime;
 
    private DataScheduleService dataScheduleService= SpringUtil.getBean(DataScheduleService.class);
 
    private DynamicDataSourceService dataSourceService= SpringUtil.getBean(DynamicDataSourceService.class);
 
    @Override
    public void run() {
        compile();
    }
 
    /**
     * 編譯
     */
    public void compile() {
        try {
            //取出一個(gè)設(shè)備
            TaskProcessor taskProcessor = TaskQueue.taskProcessors.take();
            //取出一個(gè)任務(wù)
            CompileTask compileTask = TaskQueue.waitTasks.take();
            //任務(wù)和設(shè)備綁定
            compileTask.setTaskProcessor(taskProcessor);
            //放入
            TaskQueue.executeTasks.put(compileTask);
            System.out.println(DataScheduleEnum.DEAL_WITH.getName()+" "+userId);
            //切換用戶數(shù)據(jù)源
            dataSourceService.switchDataSource(userId);
            //添加進(jìn)度
            dataScheduleService.addSchedule(clazzId, DataScheduleEnum.DEAL_WITH.getState());
        } catch (InterruptedException e) {
            System.err.println( e.getMessage());
        }
    }
 
}

import lombok.Data;
 
import java.util.Date;
 
@Data
public class TaskProcessor {
 
    /**
     * 釋放
     */
    public  static Boolean release(CompileTask task)  {
        Boolean flag=false;
        Thread thread=task.getThread();
        synchronized (thread) {
            try {
                if(null!=task.getTaskProcessor()){
                    TaskQueue.taskProcessors.put(task.getTaskProcessor());
                    TaskQueue.executeTasks.remove(task);
                    task.setEndTime(new Date());
                    long intervalMilli = task.getEndTime().getTime() - task.getStartTime().getTime();
                    flag=true;
                    System.out.println("用戶"+task.getClazzId()+"耗時(shí)"+intervalMilli+"ms");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return flag;
        }
    }
 
}

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springblade.core.tool.api.R;
import org.springblade.gis.multithread.TaskProcessor;
import org.springblade.gis.multithread.TaskQueue;
import org.springblade.gis.multithread.CompileTask;
import org.springframework.web.bind.annotation.*;
 
import java.util.Date;
 
 
@RestController
@RequestMapping("task")
@Api(value = "數(shù)據(jù)編譯任務(wù)", tags = "數(shù)據(jù)編譯任務(wù)")
public class CompileTaskController {
 
    @ApiOperation(value = "添加等待請(qǐng)求 @author Tarzan Liu")
    @PostMapping("compile/{clazzId}")
    public R<Integer> compile(@PathVariable("clazzId") Long clazzId) {
        CompileTask checkUser=TaskQueue.getWaitTask(clazzId);
        if(checkUser!=null){
            return  R.fail("已經(jīng)正在排隊(duì)！");
        }
        checkUser=TaskQueue.getExecuteTask(clazzId);
        if(checkUser!=null){
            return  R.fail("正在執(zhí)行編譯！");
        }
        //獲取當(dāng)前的線程
        Thread thread=Thread.currentThread();
        //創(chuàng)建當(dāng)前的用戶請(qǐng)求對(duì)象
        CompileTask compileTask =new CompileTask();
        compileTask.setThread(thread);
        compileTask.setClazzId(clazzId);
        compileTask.setStartTime(new Date());
        //將當(dāng)前用戶請(qǐng)求對(duì)象放入隊(duì)列中
        try {
            TaskQueue.waitTasks.put(compileTask);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        TaskQueue.exec.execute(compileTask);
        return R.data(TaskQueue.waitTasks.size()-1);
    }
 
    @ApiOperation(value = "查詢當(dāng)前任務(wù)前還有多少任務(wù)等待 @author Tarzan Liu")
    @PostMapping("sort/{clazzId}")
    public R<Integer> sort(@PathVariable("clazzId") Long clazzId) {
        return R.data(TaskQueue.getSort(clazzId));
    }
 
    @ApiOperation(value = "查詢當(dāng)前任務(wù)預(yù)估時(shí)長(zhǎng) @author Tarzan Liu")
    @PostMapping("estimate/time/{clazzId}")
    public R<Integer> estimatedTime(@PathVariable("clazzId") Long clazzId) {
        return R.data(TaskQueue.estimatedTime(clazzId));
    }
 
    @ApiOperation(value = "任務(wù)釋放 @author Tarzan Liu")
    @PostMapping("release/{clazzId}")
    public R<Boolean> release(@PathVariable("clazzId") Long clazzId) {
        CompileTask task=TaskQueue.getExecuteTask(clazzId);
        if(task==null){
            return  R.fail("資源釋放異常");
        }
        return R.status(TaskProcessor.release(task));
    }
 
    @ApiOperation(value = "執(zhí)行 @author Tarzan Liu")
    @PostMapping("exec")
    public R exec() {
        Long start=System.currentTimeMillis();
        for (Long i = 1L; i < 100; i++) {
            compile(i);
        }
        System.out.println("消耗時(shí)間："+(System.currentTimeMillis()-start)+"ms");
        return R.status(true);
    }
}