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

JVM優(yōu)先級線程池做任務(wù)隊列的實現(xiàn)方法

 更新時間:2020年08月06日 10:54:15   作者:Fururur  
這篇文章主要介紹了JVM優(yōu)先級線程池做任務(wù)隊列的實現(xiàn)方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

前言

我們都知道 web 服務(wù)的工作大多是接受 http 請求,并返回處理后的結(jié)果。服務(wù)器接受的每一個請求又可以看是一個任務(wù)。一般而言這些請求任務(wù)會根據(jù)請求的先后有序處理,如果請求任務(wù)的處理比較耗時,往往就需要排隊了。而同時不同的任務(wù)直接可能會存在一些優(yōu)先級的變化,這時候就需要引入任務(wù)隊列并進行管理了。可以做任務(wù)隊列的東西有很多,Java 自帶的線程池,以及其他的消息中間件都可以。

同步與異步

這個問題在之前已經(jīng)提過很多次了,有些任務(wù)是需要請求后立即返回結(jié)果的,而有的則不需要。設(shè)想一下你下單購物的場景,付完錢后,系統(tǒng)只需要返回一個支付成功即可,后續(xù)的積分增加、優(yōu)惠券發(fā)放、安排發(fā)貨等等業(yè)務(wù)都不需要實時返回給用戶的,這些就是異步的任務(wù)。大量的異步任務(wù)到達我們部署的服務(wù)上,由于處理效率的瓶頸,無法達到實時處理,因此與需要用隊列將他們暫時保存起來,排隊處理。

線程池

在 Java 中提到隊列,我們除了想到基本的數(shù)據(jù)結(jié)構(gòu)之外,應(yīng)該還有線程池。線程池自帶一套機制可以實現(xiàn)任務(wù)的排隊和執(zhí)行,可以滿足單點環(huán)境下絕大多數(shù)異步化的場景。下面是典型的一個處理流程:

// 注入合適類型的線程池
@Autowired
private final ThreadPoolExecutor asyncPool;
@RequestMapping(value = "/async/someOperate", method = RequestMethod.POST)
public RestResult someOperate(HttpServletRequest request, String params,String callbackUrl {
  // 接受請求后 submit 到線程池排隊處理
  asyncPool.submit(new Task(params,callbackUrl);
  return new RestResult(ResultCode.SUCCESS.getCode(), null) {{
    setMsg("successful!" + prop.getShowMsg());
  }};
}

// 異步任務(wù)處理
@Slf4j
public class Task extends Callable<RestResult> {
  private String params;
  private String callbackUrl;
  private final IAlgorithmService algorithmService = SpringUtil.getBean(IAlgorithmServiceImpl.class);
  private final ServiceUtils serviceUtils = SpringUtil.getBean(ServiceUtils.class);
  public ImageTask(String params,String callbackUrl) {
    this.params = params;
    this.callbackUrl = callbackUrl;
  }

  @Override
  public RestResult call() {
    try {
      // 業(yè)務(wù)處理
      CarDamageResult result = algorithmService.someOperate(this.params);
      // 回調(diào)
      return serviceUtils.callback(this.callbackUrl, this.caseNum, ResultCode.SUCCESS.getCode(), result, this.isAsync);
    } catch (ServiceException e) {
      return serviceUtils.callback(this.callbackUrl, this.caseNum, e.getCode(), null, this.isAsync);
    }
  }
}

對于線程池這里就不具體展開講了,僅僅簡單理了下具體的流程:

  1. 收到請求后,參數(shù)校驗后傳入線程池排隊。
  2. 返回結(jié)果:“請求成功,正在處理”。
  3. 任務(wù)排到后由相應(yīng)的線程處理,處理完后進行接口回調(diào)。

上面的例子描述了一個生產(chǎn)速度遠遠大于消費速度的模型,普通面向數(shù)據(jù)庫開發(fā)的企業(yè)級應(yīng)用,由于數(shù)據(jù)庫的連接池開發(fā)的連接數(shù)較大,一般不需要這樣通過線程池來處理,而一些 GPU 密集型的應(yīng)用場景,由于顯存的瓶頸導致消費速度慢時,就需要隊列來作出調(diào)整了。

帶優(yōu)先級的線程池

更復雜的,例如考慮到任務(wù)的優(yōu)先級,還需要對線程池進行重寫,通過 PriorityBlockingQueue 來替換默認的阻塞隊列。直接上代碼。

import lombok.Data;

import java.util.concurrent.Callable;

/**
 * @author Fururur
 * @create 2020-01-14-10:37
 */
@Data
public abstract class PriorityCallable<T> implements Callable<T> {
  private int priority;
}

import lombok.Getter;
import lombok.Setter;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 優(yōu)先級線程池的實現(xiàn)
 *
 * @author Fururur
 * @create 2019-07-23-10:19
 */
public class PriorityThreadPoolExecutor extends ThreadPoolExecutor {

  private ThreadLocal<Integer> local = ThreadLocal.withInitial(() -> 0);

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue());
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory);
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), handler);
  }

  public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                   long keepAliveTime, TimeUnit unit, ThreadFactory threadFactory,
                   RejectedExecutionHandler handler) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory, handler);
  }

  private static PriorityBlockingQueue getWorkQueue() {
    return new PriorityBlockingQueue();
  }

  @Override
  public void execute(Runnable command) {
    int priority = local.get();
    try {
      this.execute(command, priority);
    } finally {
      local.set(0);
    }
  }

  public void execute(Runnable command, int priority) {
    super.execute(new PriorityRunnable(command, priority));
  }

  public <T> Future<T> submit(PriorityCallable<T> task) {
    local.set(task.getPriority());
    return super.submit(task);
  }

  public <T> Future<T> submit(Runnable task, T result, int priority) {
    local.set(priority);
    return super.submit(task, result);
  }

  public Future<?> submit(Runnable task, int priority) {
    local.set(priority);
    return super.submit(task);
  }

  @Getter
  @Setter
  protected static class PriorityRunnable implements Runnable, Comparable<PriorityRunnable> {
    private final static AtomicLong seq = new AtomicLong();
    private final long seqNum;
    private Runnable run;

    private int priority;

    PriorityRunnable(Runnable run, int priority) {
      seqNum = seq.getAndIncrement();
      this.run = run;
      this.priority = priority;
    }

    @Override
    public void run() {
      this.run.run();
    }

    @Override
    public int compareTo(PriorityRunnable other) {
      int res = 0;
      if (this.priority == other.priority) {
        if (other.run != this.run) {
          // ASC
          res = (seqNum < other.seqNum ? -1 : 1);
        }
      } else {
        // DESC
        res = this.priority > other.priority ? -1 : 1;
      }
      return res;
    }
  }
}

要點如下:

  1. 替換線程池默認的阻塞隊列為 PriorityBlockingQueue,響應(yīng)的傳入的線程類需要實現(xiàn) Comparable<T> 才能進行比較。
  2. PriorityBlockingQueue 的數(shù)據(jù)結(jié)構(gòu)決定了,優(yōu)先級相同的任務(wù)無法保證 FIFO,需要自己控制順序。
  3. 需要重寫線程池的 execute() 方法??催^線程池源碼的會發(fā)現(xiàn),執(zhí)行 submit(task) 方法后,都會轉(zhuǎn)化成 RunnableFuture<T> 再進一步執(zhí)行,由于傳入的 task 雖然實現(xiàn)了 Comparable<T> 到,但是內(nèi)部轉(zhuǎn)換成的 RunnableFuture<T> 并未實現(xiàn),因此直接 submit 會拋出 Caused by: java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable 這樣一個異常,所以需要重寫 execute() 方法,構(gòu)造一個 PriorityRunnable 作為中轉(zhuǎn)。

總結(jié)

JVM 線程池是實現(xiàn)異步任務(wù)隊列最簡單最原生的一種方式,本文介紹了基本的使用流程和帶有優(yōu)先隊列需求的用法。這種方法可有滿足到一些簡單的業(yè)務(wù)場景,但也存在一定的局限性:

  1. JVM 線程池是單機的,橫向擴展多個服務(wù)下做負載均衡時,就會存在多個線程池了他們是分開工作的,無法很好的統(tǒng)一和管理,不太適合分布式場景。
  2. JVM 線程池是基于內(nèi)存的,一旦服務(wù)掛了,會出現(xiàn)任務(wù)丟失的情況,可靠性低。
  3. 缺少作為任務(wù)隊列的 ack 機制,一旦任務(wù)失敗不會重新執(zhí)行,且無法很好地對線程池隊列進行監(jiān)控。

顯然簡單的 JVM 線程池是無法 handle 到負載的業(yè)務(wù)場景的,這就需要引入其他中間件了,在接下來的文章中我們會繼續(xù)探討。

參考文獻

ThreadPoolExecutor 優(yōu)先級的線程池

implementing PriorityQueue on ThreadPoolExecutor

ThreadPoolExecutor 的 PriorityBlockingQueue 類型轉(zhuǎn)化問題

大搜車異步任務(wù)隊列中間件的建設(shè)實踐

到此這篇關(guān)于JVM優(yōu)先級線程池做任務(wù)隊列的實現(xiàn)方法的文章就介紹到這了,更多相關(guān)java線程池優(yōu)先級內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家! 

相關(guān)文章

  • springboot2+mybatis多種方式實現(xiàn)多數(shù)據(jù)配置方法

    springboot2+mybatis多種方式實現(xiàn)多數(shù)據(jù)配置方法

    這篇文章主要介紹了springboot2+mybatis多種方式實現(xiàn)多數(shù)據(jù)配置方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2020-03-03
  • 到底如何設(shè)置Java線程池的大小的方法示例

    到底如何設(shè)置Java線程池的大小的方法示例

    在我們?nèi)粘I(yè)務(wù)開發(fā)過程中,或多或少都會用到并發(fā)的功能。那么并發(fā)線程池到底設(shè)置多大呢?文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2019-09-09
  • Java Web過濾器詳解

    Java Web過濾器詳解

    這篇文章主要為大家詳細介紹了Java WEB過濾器的相關(guān)資料,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2017-08-08
  • Spring的Model?和?Map的原理源碼解析

    Spring的Model?和?Map的原理源碼解析

    這篇文章主要介紹了Spring的Model?和?Map的原理解析,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下
    2022-06-06
  • RocketMQ事務(wù)消息機制詳解

    RocketMQ事務(wù)消息機制詳解

    這篇文章主要介紹了RocketMQ事務(wù)消息機制詳解,RocketMQ服務(wù)端將消息持久化之后,向發(fā)送方返回Ack確認消息已經(jīng)發(fā)送成功,由于消息為半事務(wù)消息,在未收到生產(chǎn)者對該消息的二次確認前,此消息被標記成"暫不能投遞"狀態(tài),需要的朋友可以參考下
    2024-01-01
  • Intellij搭建springmvc常見問題解決方案

    Intellij搭建springmvc常見問題解決方案

    這篇文章主要介紹了Intellij搭建springmvc常見問題解決方案,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下
    2020-10-10
  • Java多線程常見案例分析線程池與單例模式及阻塞隊列

    Java多線程常見案例分析線程池與單例模式及阻塞隊列

    這篇文章主要介紹了多線程的常見案例,線程池(重點)、單例模式、阻塞隊列,本文通過圖文實例相結(jié)合給大家介紹的非常詳細,需要的朋友可以參考下
    2022-05-05
  • 解決IDEA刪除子模塊并重建后MAVEN無法識別的問題

    解決IDEA刪除子模塊并重建后MAVEN無法識別的問題

    這篇文章主要介紹了解決IDEA刪除子模塊并重建后MAVEN無法識別的問題,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2021-02-02
  • java中的switch case語句使用詳解

    java中的switch case語句使用詳解

    這篇文章主要介紹了java中的switch case語句使用詳解,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
    2020-07-07
  • CentOS8.2安裝Java 14.0.2的教程詳解

    CentOS8.2安裝Java 14.0.2的教程詳解

    這篇文章主要介紹了CentOS8.2安裝Java 14.0.2的詳細教程,本文給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下
    2020-12-12

最新評論

普兰县| 施秉县| 文山县| 开远市| 磴口县| 平南县| 延庆县| 丰都县| 阿尔山市| 陈巴尔虎旗| 曲阳县| 邵武市| 会东县| 蛟河市| 界首市| 东安县| 安徽省| 宜宾市| 滕州市| 鲜城| 宁陕县| 平武县| 北流市| 通州市| 灌南县| 井冈山市| 岳池县| 马关县| 武定县| 万荣县| 胶南市| 定结县| 织金县| 辽源市| 乌拉特中旗| 龙江县| 余庆县| 敦化市| 枞阳县| 德江县| 平塘县|