前言
在工作中，很多高并发的场景中，我们会用到队列来实现大量的任务请求。当任务需要某些特殊资源的时候，我们还需要合理的分配资源，让队列中的任务高效且有序完成任务。熟悉分布式的话，应该了解yarn的任务调度算法。本文主要用java实现一个FIFO（先进先出调度器），这也是常见的一种调度方式。
FIFO任务调度器架构
主要实现的逻辑可以归纳为：
1、任务队列主要是单队列，所有任务按照顺序进入队列后，也会按照顺序执行。
2、如果任务无法获得资源，则将任务塞回队列原位置。
示例代码
Maven依赖如下：
<dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-all</artifactId> <version>5.5.2</version> </dependency>
具体的原理就不细说了，通过代码我们看看FIFO任务调度策略是什么玩的吧。下面的代码也可以作为参考。我们会使用到一个双向阻塞队列LinkedBlockingDeque。后面的代码说明会提到。
package ai.guiji.csdn.dispatch; import cn.hutool.core.thread.ThreadUtil; import lombok.Builder; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import java.util.Random; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; /** * @Program: csdn @ClassName: FIFODemo @Author: 剑客阿良_ALiang @Date: 2021-12-24 21:21 @Description: * fifo队列 @Version: V1.0 */ @Slf4j public class FIFODemo { private static final LinkedBlockingDeque<Task> TASK_QUEUE = new LinkedBlockingDeque<>(); private static final ConcurrentHashMap<Integer, LinkedBlockingQueue<Resource>> RESOURCE_MAP = new ConcurrentHashMap<>(); private static final ExecutorService TASK_POOL = new ThreadPoolExecutor( 8, 16, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new CustomizableThreadFactory("TASK-THREAD-"), new ThreadPoolExecutor.AbortPolicy()); private static final ScheduledExecutorService ENGINE_POOL = Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("ENGINE-")); private static final AtomicInteger CODE_BUILDER = new AtomicInteger(0); @Data @Builder private static class Resource { private Integer rId; private Type type; } @Data @Builder private static class Task implements Runnable { private Integer tId; private Runnable work; private Type type; private Resource resource; @Override public void run() { log.info("[{}]任务，使用资源编号:[{}]", tId, resource.getRId()); try { work.run(); } catch (Exception exception) { exception.printStackTrace(); } finally { log.info("[{}]任务结束，回归资源", tId); returnResource(resource); } } } private enum Type { /** 资源类型 */ A("A资源", 1), B("B资源", 2), C("C资源", 3); private final String desc; private final Integer code; Type(String desc, Integer code) { this.desc = desc; this.code = code; } public String getDesc() { return desc; } public Integer getCode() { return code; } } public static void initResource() { Random random = new Random(); int aCount = random.nextInt(10) + 1; int bCount = random.nextInt(10) + 1; int cCount = random.nextInt(10) + 1; RESOURCE_MAP.put(Type.A.getCode(), new LinkedBlockingQueue<>()); RESOURCE_MAP.put(Type.B.getCode(), new LinkedBlockingQueue<>()); RESOURCE_MAP.put(Type.C.getCode(), new LinkedBlockingQueue<>()); IntStream.rangeClosed(1, aCount) .forEach( a -> RESOURCE_MAP .get(Type.A.getCode()) .add(Resource.builder().rId(a).type(Type.A).build())); IntStream.rangeClosed(1, bCount) .forEach( a -> RESOURCE_MAP .get(Type.B.getCode()) .add(Resource.builder().rId(a).type(Type.B).build())); IntStream.rangeClosed(1, cCount) .forEach( a -> RESOURCE_MAP .get(Type.C.getCode()) .add(Resource.builder().rId(a).type(Type.C).build())); log.info("初始化资源A数量:{},资源B数量:{},资源C数量:{}", aCount, bCount, cCount); } public static Resource extractResource(Type type) { return RESOURCE_MAP.get(type.getCode()).poll(); } public static void returnResource(Resource resource) { log.info("开始归还资源，rId:{},资源类型:{}", resource.getRId(), resource.getType().getDesc()); RESOURCE_MAP.get(resource.getType().code).add(resource); log.info("归还资源完成，rId:{},资源类型:{}", resource.getRId(), resource.getType().getDesc()); } public static void enginDo() { ENGINE_POOL.scheduleAtFixedRate( () -> { Task task = TASK_QUEUE.poll(); if (task == null) { log.info("任务队列为空，无需要执行的任务"); } else { Resource resource = extractResource(task.getType()); if (resource == null) { log.info("[{}]任务无法获取[{}]，返回队列", task.getTId(), task.getType().getDesc()); TASK_QUEUE.addFirst(task); } else { task.setResource(resource); TASK_POOL.submit(task); } } }, 0, 1, TimeUnit.SECONDS); } public static void addTask(Runnable runnable, Type type) { Integer tId = CODE_BUILDER.incrementAndGet(); Task task = Task.builder().tId(tId).type(type).work(runnable).build(); log.info("提交任务[{}]到任务队列", tId); TASK_QUEUE.add(task); } public static void main(String[] args) { initResource(); enginDo(); Random random = new Random(); ThreadUtil.sleep(5000); IntStream.range(0, 10) .forEach( a -> addTask(() -> ThreadUtil.sleep(random.nextInt(10) + 1, TimeUnit.SECONDS), Type.A)); IntStream.range(0, 10) .forEach( a -> addTask(() -> ThreadUtil.sleep(random.nextInt(10) + 1, TimeUnit.SECONDS), Type.B)); IntStream.range(0, 10) .forEach( a -> addTask(() -> ThreadUtil.sleep(random.nextInt(10) + 1, TimeUnit.SECONDS), Type.C)); } }
代码说明：
1、首先我们构造了任务队列，使用的是LinkedBlockingDeque，使用双向队列的原因是如果任务无法获取资源，还需要塞到队首，保证任务的有序性。
2、使用ConcurrentHashMap作为资源映射表，为了保证资源队列使用的均衡性，一旦使用完成的资源会塞到对应资源的队尾处。
3、其中实现了添加任务、提取资源、回归资源几个方法。
4、initResource方法可以初始化资源队列，这里面只是简单的随机了几个资源到A、B、C三种资源，塞入各类别队列。
5、任务私有类有自己的任务标识以及执行完后调用回归资源方法。
6、main方法中会分别提交需要3中资源的10个任务，看看调度情况。
执行结果
我们可以通过结果发现任务有序调度，使用完任务后回归队列。
总结
在工作中如果有用到的话，可以参考参考，本文主要是分享。
分享：
when you really think about it？ one of the hardest lessons in life is letting go. whether it`s guilt anger love or loss. Change is never easy .We fight to hold on, and we fight to let go. ——我也不知道来自哪，偶然看到的一句话。
如果本文对你有帮助的话，点个赞吧，谢谢！
