JVM优先级线程池做任务队列的实现方法

// 注入合适类型的线程池
@Autowired
private final ThreadPoolExecutor asyncPool;
@RequestMapping(value = "/async/someOperate", method = RequestMethod.POST)
public RestResult someOperate(HttpServletRequest request, String params,String callbackUrl {
  // 接受请求后 submit 到线程池排队处理
  asyncPool.submit(new Task(params,callbackUrl);
  return new RestResult(ResultCode.SUCCESS.getCode(), null) {{
    setMsg("successful!" + prop.getShowMsg());
  }};
}

// 异步任务处理
@Slf4j
public class Task extends Callable<RestResult> {
  private String params;
  private String callbackUrl;
  private final IAlgorithmService algorithmService = SpringUtil.getBean(IAlgorithmServiceImpl.class);
  private final ServiceUtils serviceUtils = SpringUtil.getBean(ServiceUtils.class);
  public ImageTask(String params,String callbackUrl) {
    this.params = params;
    this.callbackUrl = callbackUrl;
  }

  @Override
  public RestResult call() {
    try {
      // 业务处理
      CarDamageResult result = algorithmService.someOperate(this.params);
      // 回调
      return serviceUtils.callback(this.callbackUrl, this.caseNum, ResultCode.SUCCESS.getCode(), result, this.isAsync);
    } catch (ServiceException e) {
      return serviceUtils.callback(this.callbackUrl, this.caseNum, e.getCode(), null, this.isAsync);
    }
  }
}

import lombok.Data;

import java.util.concurrent.Callable;

/**
 * @author Fururur
 * @create 2020-01-14-10:37
 */
@Data
public abstract class PriorityCallable<T> implements Callable<T> {
  private int priority;
}

import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 优先级线程池的实现
 *
 * @author Fururur
 * @create 2019-07-23-10:19
 */
public class PriorityThreadPoolExecutor extends ThreadPoolExecutor {

  private ThreadLocal<Integer> local = ThreadLocal.withInitial(() -> 0);

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue());
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory);
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), handler);
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory,
                   RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory, handler);
  }

  private static PriorityBlockingQueue getWorkQueue() {
    return new PriorityBlockingQueue();
  }

  @Override
  public void execute(Runnable command) {
    int priority = local.get();
    try {
      this.execute(command, priority);
    } finally {
      local.set(0);
    }
  }

  public void execute(Runnable command, int priority) {
    super.execute(new PriorityRunnable(command, priority));
  }

  public <T> Future<T> submit(PriorityCallable<T> task) {
    local.set(task.getPriority());
    return super.submit(task);
  }

  public <T> Future<T> submit(Runnable task, T result, int priority) {
    local.set(priority);
    return super.submit(task, result);
  }

  public Future<?> submit(Runnable task, int priority) {
    local.set(priority);
    return super.submit(task);
  }

  @Getter
  @Setter
  protected static class PriorityRunnable implements Runnable, Comparable<PriorityRunnable> {
    private final static AtomicLong seq = new AtomicLong();
    private final long seqNum;
    private Runnable run;

    private int priority;

    PriorityRunnable(Runnable run, int priority) {
      seqNum = seq.getAndIncrement();
      this.run = run;
      this.priority = priority;
    }

    @Override
    public void run() {
      this.run.run();
    }

    @Override
    public int compareTo(PriorityRunnable other) {
      int res = 0;
      if (this.priority == other.priority) {
        if (other.run != this.run) {
          // ASC
          res = (seqNum < other.seqNum ? -1 : 1);
        }
      } else {
        // DESC
        res = this.priority > other.priority ? -1 : 1;
      }
      return res;
    }
  }
}