最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Java實(shí)現(xiàn)FIFO任務(wù)調(diào)度隊(duì)列策略

 更新時(shí)間:2021年12月27日 08:37:00   作者:劍客阿良_ALiang  
在工作中,很多高并發(fā)的場(chǎng)景中,我們會(huì)用到隊(duì)列來(lái)實(shí)現(xiàn)大量的任務(wù)請(qǐng)求。當(dāng)任務(wù)需要某些特殊資源的時(shí)候,我們還需要合理的分配資源,讓隊(duì)列中的任務(wù)高效且有序完成任務(wù)。本文將為大家介紹通過(guò)java實(shí)現(xiàn)FIFO任務(wù)調(diào)度,需要的可以參考一下

前言

在工作中,很多高并發(fā)的場(chǎng)景中,我們會(huì)用到隊(duì)列來(lái)實(shí)現(xiàn)大量的任務(wù)請(qǐng)求。當(dāng)任務(wù)需要某些特殊資源的時(shí)候,我們還需要合理的分配資源,讓隊(duì)列中的任務(wù)高效且有序完成任務(wù)。熟悉分布式的話(huà),應(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依賴(lài)如下:

      	<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)文章

  • 淺談Spring事務(wù)傳播行為實(shí)戰(zhàn)

    淺談Spring事務(wù)傳播行為實(shí)戰(zhàn)

    這篇文章主要介紹了淺談Spring事務(wù)傳播行為實(shí)戰(zhàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
    2019-09-09
  • Spring Boot 注解方式自定義Endpoint詳解

    Spring Boot 注解方式自定義Endpoint詳解

    這篇文章主要介紹了Spring Boot注解方式自定義Endpoint詳解,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2021-09-09
  • Java五子棋簡(jiǎn)單實(shí)現(xiàn)代碼舉例

    Java五子棋簡(jiǎn)單實(shí)現(xiàn)代碼舉例

    Java五子棋游戲是一種經(jīng)典的兩人對(duì)戰(zhàn)棋類(lèi)游戲,它基于簡(jiǎn)單的規(guī)則,即任何一方的棋子在棋盤(pán)上形成連續(xù)的五個(gè),無(wú)論是橫、豎還是斜線(xiàn),都將獲勝,這篇文章主要介紹了Java五子棋實(shí)現(xiàn)的相關(guān)資料,需要的朋友可以參考下
    2024-10-10
  • PostMan傳@RequestParam修飾的數(shù)組方式

    PostMan傳@RequestParam修飾的數(shù)組方式

    這篇文章主要介紹了PostMan傳@RequestParam修飾的數(shù)組方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教
    2021-08-08
  • SpringBoot實(shí)現(xiàn)文件斷點(diǎn)續(xù)傳功能詳解

    SpringBoot實(shí)現(xiàn)文件斷點(diǎn)續(xù)傳功能詳解

    在處理大文件傳輸或網(wǎng)絡(luò)不穩(wěn)定的情況下,文件斷點(diǎn)續(xù)傳功能顯得尤為重要,本文將詳細(xì)介紹如何使用Spring Boot實(shí)現(xiàn)文件的斷點(diǎn)續(xù)傳功能,需要的可以了解下
    2025-04-04
  • 一文掌握J(rèn)ava開(kāi)發(fā)工具M(jìn)aven(簡(jiǎn)單上手)

    一文掌握J(rèn)ava開(kāi)發(fā)工具M(jìn)aven(簡(jiǎn)單上手)

    掌握maven的相關(guān)知識(shí)是Java開(kāi)發(fā)必備的技能,今天通過(guò)本文從入門(mén)安裝開(kāi)始,逐步深入講解maven的相關(guān)知識(shí),包括maven的安裝到簡(jiǎn)單上手maven項(xiàng)目開(kāi)發(fā),感興趣的朋友跟隨小編一起看看吧
    2021-06-06
  • Java EE項(xiàng)目中的異常處理總結(jié)(一篇不得不看的文章)

    Java EE項(xiàng)目中的異常處理總結(jié)(一篇不得不看的文章)

    什么是異常?運(yùn)行時(shí)發(fā)生的可被捕獲和處理的錯(cuò)誤。這篇文章主要介紹了Java EE項(xiàng)目中的異常處理總結(jié),有需要的可以了解一下。
    2016-11-11
  • java pdf加水印的方法

    java pdf加水印的方法

    這篇文章主要為大家詳細(xì)介紹了java pdf加水印的方法,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下
    2019-08-08
  • linux環(huán)境下java程序打包成簡(jiǎn)單的hello world輸出jar包示例

    linux環(huán)境下java程序打包成簡(jiǎn)單的hello world輸出jar包示例

    這篇文章主要介紹了linux環(huán)境下java程序打包成簡(jiǎn)單的hello world輸出jar包,結(jié)合簡(jiǎn)單hello world輸出程序示例分析了Linux環(huán)境下的java可執(zhí)行jar包文件的生成相關(guān)操作技巧,需要的朋友可以參考下
    2019-11-11
  • 淺析Java中的異常處理機(jī)制

    淺析Java中的異常處理機(jī)制

    這篇文章主要介紹了Java中的異常處理機(jī)制的相關(guān)資料,幫助大家更好的理解和學(xué)習(xí)Java,感興趣的朋友可以了解下
    2020-11-11

最新評(píng)論

马山县| 汾阳市| 包头市| 积石山| 广西| 宁陵县| 黔东| 莱芜市| 香格里拉县| 漾濞| 衡水市| 屯昌县| 墨玉县| 台安县| 德惠市| 临颍县| 城口县| 马边| 德庆县| 昂仁县| 盐源县| 阿坝县| 彭泽县| 璧山县| 封丘县| 辽源市| 镇平县| 安塞县| 盐边县| 海盐县| 米脂县| 梅州市| 若尔盖县| 罗城| 通化县| 开化县| 奎屯市| 抚远县| 隆安县| 大宁县| 固始县|