Java实现FIFO任务调度队列策略

Java实现FIFO任务调度队列策略

目录前言FIFO任务调度器架构示例代码

前言

在工作中，很多高并发的场景中，我们会用到队列来实现大量的任务请求。当任务需要某些特殊资源的时候，我们还需要合理的分配资源，让队列中的任务高效且有序完成任务。熟悉分布式的话，应该了解yarn的任务NzeQlbTz调度算法。本文主要用java实现一个FIFO（先进先出调度器），这也是常见的一种调度方式。

FIFO任务调度器架构

主要实现的逻辑可以归纳为：

1、任务队列主要是单队列，所有任务按照顺序进入队列后，也会按照顺序执行。

http://

2、如果任务无法获得资源，则将任务塞回队列原位置。

示例代码

Maven依赖如下：

org.projectlombok

lombok

true

cn.hutool

hutool-all

5.5.2

具体的原理就不细说了，通过代码我们看看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_QUEUE = new LinkedBlockingDeque<>();

private static final ConcurrentHashMap> 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个任务，看看调度情况。

执行结果

我们可以通过结果发现任务有序调度，使用完任务后回归队列。

以上就是Java实现FIFO任务调度队列策略的详细内容，更多关于Java FIFO任务调度的资料请关注我们其它相关文章！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。