Java實現(xiàn)FIFO任務(wù)調(diào)度隊列策略
前言
在工作中,很多高并發(fā)的場景中,我們會用到隊列來實現(xiàn)大量的任務(wù)請求。當(dāng)任務(wù)需要某些特殊資源的時候,我們還需要合理的分配資源,讓隊列中的任務(wù)高效且有序完成任務(wù)。熟悉分布式的話,應(yīng)該了解yarn的任務(wù)調(diào)度算法。本文主要用java實現(xiàn)一個FIFO(先進先出調(diào)度器),這也是常見的一種調(diào)度方式。
FIFO任務(wù)調(diào)度器架構(gòu)
主要實現(xiàn)的邏輯可以歸納為:
1、任務(wù)隊列主要是單隊列,所有任務(wù)按照順序進入隊列后,也會按照順序執(zhí)行。
2、如果任務(wù)無法獲得資源,則將任務(wù)塞回隊列原位置。
示例代碼
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ì)說了,通過代碼我們看看FIFO任務(wù)調(diào)度策略是什么玩的吧。下面的代碼也可以作為參考。我們會使用到一個雙向阻塞隊列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("[{}]任務(wù),使用資源編號:[{}]", tId, resource.getRId()); try { work.run(); } catch (Exception exception) { exception.printStackTrace(); } finally { log.info("[{}]任務(wù)結(jié)束,回歸資源", 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數(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("開始?xì)w還資源,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("任務(wù)隊列為空,無需要執(zhí)行的任務(wù)"); } else { Resource resource = extractResource(task.getType()); if (resource == null) { log.info("[{}]任務(wù)無法獲取[{}],返回隊列", 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ù)隊列", 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、首先我們構(gòu)造了任務(wù)隊列,使用的是LinkedBlockingDeque,使用雙向隊列的原因是如果任務(wù)無法獲取資源,還需要塞到隊首,保證任務(wù)的有序性。
2、使用ConcurrentHashMap作為資源映射表,為了保證資源隊列使用的均衡性,一旦使用完成的資源會塞到對應(yīng)資源的隊尾處。
3、其中實現(xiàn)了添加任務(wù)、提取資源、回歸資源幾個方法。
4、initResource方法可以初始化資源隊列,這里面只是簡單的隨機了幾個資源到A、B、C三種資源,塞入各類別隊列。
5、任務(wù)私有類有自己的任務(wù)標(biāo)識以及執(zhí)行完后調(diào)用回歸資源方法。
6、main方法中會分別提交需要3中資源的10個任務(wù),看看調(diào)度情況。
執(zhí)行結(jié)果
我們可以通過結(jié)果發(fā)現(xiàn)任務(wù)有序調(diào)度,使用完任務(wù)后回歸隊列。?
以上就是Java實現(xiàn)FIFO任務(wù)調(diào)度隊列策略的詳細(xì)內(nèi)容,更多關(guān)于Java FIFO任務(wù)調(diào)度的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
java?list和map切割分段的實現(xiàn)及多線程應(yīng)用案例
這篇文章主要為大家介紹了java?list和map切割分段的實現(xiàn)及多線程應(yīng)用案例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-12-12Java使用MulticastSocket實現(xiàn)群聊應(yīng)用程序
這篇文章主要為大家詳細(xì)介紹了Java使用MulticastSocket實現(xiàn)群聊應(yīng)用程序,文中示例代碼介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們可以參考一下2022-05-05SpringMVC中@RequestMapping注解的實現(xiàn)
RequestMapping是一個用來處理請求地址映射的注解,本文主要介紹了SpringMVC中@RequestMapping注解的實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2024-01-01Java?實現(xiàn)訂單未支付超時自動取消功能(京東商城為例)
本文以京東網(wǎng)上商城為例,給大家介紹商品在下單后沒有支付的情況下,超時自動取消功能,超過24小時,就會自動取消訂單,下面使用 Java 定時器實現(xiàn)超時取消訂單功能,感興趣的朋友一起看看吧2022-01-01Java之maven打完jar包之后將jar包放到指定位置匯總
這篇文章主要介紹了Java之maven打完jar包之后將jar包放到指定位置匯總,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2023-04-04