下載任務執行器簡單設計

下載任務類設計

/**
 * 下載任務執行器
 * @author Shixy
 */
@Slf4j
public class Downloader implements Runnable {

    private String filePath;
    private Callable<String> callable;
    private DownloadTask downloadTask;
    private final DownloadTaskService downloadTaskService;

    private Downloader(DownloadTaskService downloadTaskService) {
        this.downloadTaskService = downloadTaskService;
    }

    /**
     * 創建一個任務執行器
     * @param downloadTaskService 下載任務Service
     * @return 任務執行器
     */
    public static Downloader getInstance(DownloadTaskService downloadTaskService) {
        return new Downloader(downloadTaskService);
    }

    /**
     * 創建任務
     * @param taskName 任務名稱
     * @param type 任務類型
     * @return 任務執行器
     */
    public Downloader create(String taskName, DownloadTypeEnum type) {
        DownloadTask downloadTask = new DownloadTask();
        downloadTask.setUserId(SecurityUtil.getCurrentUserId());
        downloadTask.setDownloadName(taskName);
        downloadTask.setDownloadType(type.getType());
        downloadTask.setDownloadStatus(DownloadStatusEnum.STATUS_GENERATING.getType());
        downloadTask.setDownloadDescription(type.getName());
        downloadTaskService.save(downloadTask);
        this.downloadTask = downloadTask;
        return this;
    }

    /**
     * 執行任務
     * @param callable 有返回值的線程
     * @return 任務執行器
     */
    public Downloader execute(Callable<String> callable) {
        this.callable = callable;
        return this;
    }

    /**
     * 完成任務
     * @return 下載路徑
     */
    public String finish() {
        this.downloadTask.setDownloadPath(this.filePath);
        this.downloadTask.setDownloadStatus(DownloadStatusEnum.STATUS_SUCCESS.getType());
        downloadTaskService.updateById(downloadTask);
        DownloaderFactory.getInstance().remove(this);
        return this.filePath;
    }

    /**
     * 任務失敗
     */
    private void taskFail() {
        downloadTask.setDownloadStatus(DownloadStatusEnum.STATUS_FAIL.getType());
        downloadTaskService.updateById(downloadTask);
        DownloaderFactory.getInstance().remove(this);
    }

    @Override
    public void run() {
        try {
            this.filePath = callable.call();
            finish();
        } catch (Exception e) {
            log.error("下載任務執行中異常 Error:", e);
            taskFail();
            throw new BadRequestException("任務執行中異常");
        }
    }

    /**
     * 啟動線程
     */
    public void start() {
        DownloaderFactory.getInstance().push(this);
    }

    public String getName() {
        if (null != downloadTask) {
            return downloadTask.getDownloadName();
        }
        return null;
    }
}

下載工廠類設計

/**
 * @Description 下載工廠類
 * @Date 2022/6/6 16:11
 * @Author Shixy
 */
@Slf4j
public class DownloaderFactory {

    /**
     * 單例
     */
    private static DownloaderFactory downloaderFactory = new DownloaderFactory();

    /**
     * 構造方法私有化
     */
    private DownloaderFactory() {}

    /**
     * 最大等待隊列
     */
    private static final int MAX_WAIT_SIZE = 100;

    /**
     * 最大執行任務數
     */
    private static final int MAX_EXEC_SIZE = 10;

    /**
     * 是否正在運行
     */
    private static boolean executing = false;

    /**
     * 等待下載隊列
     */
    private static ConcurrentLinkedDeque<Downloader> downloaderDeque = new ConcurrentLinkedDeque();

    /**
     * 執行中的隊列
     */
    private static ConcurrentLinkedDeque<Downloader> executingDeque = new ConcurrentLinkedDeque<>();

    /**
     * 單例工廠
     * @return 生產工廠
     */
    public static DownloaderFactory getInstance() {
        return downloaderFactory;
    }

    /**
     * 推送下載任務至等待隊列
     * @param downloader 任務
     */
    public void push(Downloader downloader) {
        if (MAX_WAIT_SIZE <= downloaderDeque.size()) {
            throw new BadRequestException("當前系統繁忙，請稍候再試！");
        }
        log.info("任務已進入隊列--->{}", downloader.getName());
        downloaderDeque.push(downloader);
    }

    /**
     * 激活生產線
     */
    public void start() {
        if (!executing) {
            log.info("執行器正在工作--->");
            AsyncManager.instance().execute(() -> execute());
        }
        if (downloaderDeque.size() == 0) {
            log.info("一直沒活來，眯一會zzZ");
            try {
                Thread.sleep(60000L);
            } catch (InterruptedException e) {
                log.info("工廠出事了，檢查一下");
            }
        }
        start();
    }

    /**
     * 生產線
     */
    private void execute() {
        executing = true;
        try {
            while (true) {
                if (MAX_EXEC_SIZE <= executingDeque.size()) {
                    log.info("流水線已經爆滿了，等會再來吧");
                    continue;
                }
                if (downloaderDeque.size() > 0) {
                    Downloader downloader = downloaderDeque.pop();
                    log.info("老鐵，來活了--->{}", downloader.getName());
                    executingDeque.push(downloader);
                    try {
                        AsyncManager.instance().execute(downloader);
                    } catch (Exception e) {
                        log.info("生產失敗咯，換下一個吧：", e);
                        remove(downloader);
                        continue;
                    }
                }
            }
        } catch (Exception e) {
            log.error("生產線著火了：", e);
            executing = false;
        }
    }

    /**
     * 生產完成或失敗了，扔出隊列吧
     * @param downloader 下載任務
     */
    public void remove(Downloader downloader) {
        executingDeque.remove(downloader);
    }

}

系統問題

• 下載需要生成很多零散的文件
• 需要對零散的文件進行分組，排序和壓縮
• 若不加以限制系統執行的任務數量，極可能造成OOM問題從而搞垮系統

分享

• 該類有關聯的下載任務業務類需要根據自己項目的實際情況進行替換
• 最大執行隊列也需要根據伺服器的性能進行評估
• 最後歡迎各位大佬指點小弟，萬分感激