Java實(shí)現(xiàn)FIFO任務(wù)調(diào)度隊(duì)列策略
前言
在工作中,很多高并發(fā)的場(chǎng)景中,我們會(huì)用到隊(duì)列來(lái)實(shí)現(xiàn)大量的任務(wù)請(qǐng)求。當(dāng)任務(wù)需要某些特殊資源的時(shí)候,我們還需要合理的分配資源,讓隊(duì)列中的任務(wù)高效且有序完成任務(wù)。熟悉分布式的話,應(yīng)該了解yarn的任務(wù)調(diào)度算法。本文主要用java實(shí)現(xiàn)一個(gè)FIFO(先進(jìn)先出調(diào)度器),這也是常見(jiàn)的一種調(diào)度方式。
FIFO任務(wù)調(diào)度器架構(gòu)
主要實(shí)現(xiàn)的邏輯可以歸納為:
1、任務(wù)隊(duì)列主要是單隊(duì)列,所有任務(wù)按照順序進(jìn)入隊(duì)列后,也會(huì)按照順序執(zhí)行。
2、如果任務(wù)無(wú)法獲得資源,則將任務(wù)塞回隊(duì)列原位置。
示例代碼
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>
具體的原理就不細(xì)說(shuō)了,通過(guò)代碼我們看看FIFO任務(wù)調(diào)度策略是什么玩的吧。下面的代碼也可以作為參考。我們會(huì)使用到一個(gè)雙向阻塞隊(duì)列LinkedBlockingDeque。后面的代碼說(shuō)明會(huì)提到。
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隊(duì)列 @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("[{}]任務(wù),使用資源編號(hào):[{}]", tId, resource.getRId());
try {
work.run();
} catch (Exception exception) {
exception.printStackTrace();
} finally {
log.info("[{}]任務(wù)結(jié)束,回歸資源", tId);
returnResource(resource);
}
}
}
private enum Type {
/** 資源類(lèi)型 */
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數(shù)量:{},資源B數(shù)量:{},資源C數(shù)量:{}", aCount, bCount, cCount);
}
public static Resource extractResource(Type type) {
return RESOURCE_MAP.get(type.getCode()).poll();
}
public static void returnResource(Resource resource) {
log.info("開(kāi)始?xì)w還資源,rId:{},資源類(lèi)型:{}", resource.getRId(), resource.getType().getDesc());
RESOURCE_MAP.get(resource.getType().code).add(resource);
log.info("歸還資源完成,rId:{},資源類(lèi)型:{}", resource.getRId(), resource.getType().getDesc());
}
public static void enginDo() {
ENGINE_POOL.scheduleAtFixedRate(
() -> {
Task task = TASK_QUEUE.poll();
if (task == null) {
log.info("任務(wù)隊(duì)列為空,無(wú)需要執(zhí)行的任務(wù)");
} else {
Resource resource = extractResource(task.getType());
if (resource == null) {
log.info("[{}]任務(wù)無(wú)法獲取[{}],返回隊(duì)列", 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("提交任務(wù)[{}]到任務(wù)隊(duì)列", 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));
}
}
代碼說(shuō)明:
1、首先我們構(gòu)造了任務(wù)隊(duì)列,使用的是LinkedBlockingDeque,使用雙向隊(duì)列的原因是如果任務(wù)無(wú)法獲取資源,還需要塞到隊(duì)首,保證任務(wù)的有序性。
2、使用ConcurrentHashMap作為資源映射表,為了保證資源隊(duì)列使用的均衡性,一旦使用完成的資源會(huì)塞到對(duì)應(yīng)資源的隊(duì)尾處。
3、其中實(shí)現(xiàn)了添加任務(wù)、提取資源、回歸資源幾個(gè)方法。
4、initResource方法可以初始化資源隊(duì)列,這里面只是簡(jiǎn)單的隨機(jī)了幾個(gè)資源到A、B、C三種資源,塞入各類(lèi)別隊(duì)列。
5、任務(wù)私有類(lèi)有自己的任務(wù)標(biāo)識(shí)以及執(zhí)行完后調(diào)用回歸資源方法。
6、main方法中會(huì)分別提交需要3中資源的10個(gè)任務(wù),看看調(diào)度情況。
執(zhí)行結(jié)果



我們可以通過(guò)結(jié)果發(fā)現(xiàn)任務(wù)有序調(diào)度,使用完任務(wù)后回歸隊(duì)列。?
以上就是Java實(shí)現(xiàn)FIFO任務(wù)調(diào)度隊(duì)列策略的詳細(xì)內(nèi)容,更多關(guān)于Java FIFO任務(wù)調(diào)度的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
23種設(shè)計(jì)模式(22)java狀態(tài)模式
這篇文章主要為大家詳細(xì)介紹了23種設(shè)計(jì)模式之java狀態(tài)模式,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2018-01-01
java?list和map切割分段的實(shí)現(xiàn)及多線程應(yīng)用案例
這篇文章主要為大家介紹了java?list和map切割分段的實(shí)現(xiàn)及多線程應(yīng)用案例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-12-12
Java使用MulticastSocket實(shí)現(xiàn)群聊應(yīng)用程序
這篇文章主要為大家詳細(xì)介紹了Java使用MulticastSocket實(shí)現(xiàn)群聊應(yīng)用程序,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2022-05-05
SpringMVC中@RequestMapping注解的實(shí)現(xiàn)
RequestMapping是一個(gè)用來(lái)處理請(qǐng)求地址映射的注解,本文主要介紹了SpringMVC中@RequestMapping注解的實(shí)現(xiàn),具有一定的參考價(jià)值,感興趣的可以了解一下2024-01-01
Java?實(shí)現(xiàn)訂單未支付超時(shí)自動(dòng)取消功能(京東商城為例)
本文以京東網(wǎng)上商城為例,給大家介紹商品在下單后沒(méi)有支付的情況下,超時(shí)自動(dòng)取消功能,超過(guò)24小時(shí),就會(huì)自動(dòng)取消訂單,下面使用 Java 定時(shí)器實(shí)現(xiàn)超時(shí)取消訂單功能,感興趣的朋友一起看看吧2022-01-01
Java之maven打完jar包之后將jar包放到指定位置匯總
這篇文章主要介紹了Java之maven打完jar包之后將jar包放到指定位置匯總,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-04-04
javacv視頻抽幀的實(shí)現(xiàn)過(guò)程詳解(附代碼)
這篇文章主要介紹了javacv視頻抽幀的實(shí)現(xiàn)過(guò)程詳解(附代碼),視頻抽幀可以做一些處理,比如水印,去水印等操作,然后再合成視頻,需要的朋友可以參考下2019-07-07

