web應用中實現異步任務隊列

開發中遇到這樣一個需求。導出excel可能會超時。因此須要作成異步下載。由於下載線程會比較佔用cpu資源，可能會有一些慢sql，因此須要控制併發數。

1.創建一個堵塞鏈表隊列，用來存在任務；

2.收到用戶請求建立一個下載任務，放在任務隊列中。

3.在web.xml中設置一個監聽類，啓動一個線程定時掃描任務隊列，當有任務時執行任務。

4.在執行任務時設置FutureTask超時時間，堵塞當前線程，當前任務一直不結束時，終止任務運行，釋放資源。

代碼：

<listener>
        <listener-class>com.taobao.kelude.k3case.listenners.ExportTaskStartupListener</listener-class>
</listener>

public class ExportTaskStartupListener implements ServletContextListener {
    @Override
    public void contextDestroyed(ServletContextEvent sce) {
        sce.getServletContext().log("定時器銷燬");
    }
    @Override
    public void contextInitialized(ServletContextEvent sce) {
        sce.getServletContext().log("啓動線程池");
        sce.getServletContext().log("啓動定時器");
        sce.getServletContext().log("已經添加任務調度表");
        ExportTaskQueueThread exportTaskQueueThread = new ExportTaskQueueThread();
        exportTaskQueueThread.start();
    }
}

class ExportTaskQueueThread extends Thread {
    public void run() {
        System.out.println("init timer task");
        while (true) {
//            System.out.println("start timer task");
            try {
                if (!ExportTaskPoolManager.queue.isEmpty()) {
                    CallableTask callableTask = ExportTaskPoolManager.queue.take();
                    FutureTask futureTask = new FutureTask(callableTask);
                    Thread thread = new Thread(futureTask);
                    thread.start();
                    try {
                        futureTask.get(300, TimeUnit.SECONDS);
                    } catch (TimeoutException e) {
                        System.out.println("export failed for timeout");
                        futureTask.cancel(true);
                    }
                }
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}

 

public class ExportTaskPoolManager {
//    public static ExecutorService pool;
    public static LinkedBlockingQueue<CallableTask> queue = new LinkedBlockingQueue<CallableTask>();
}

public class ExportCaseTask extends CallableTask {
    private Logger logger = LoggerFactory.getLogger("ExportCaseTask");
    private List<Integer> priority;
    private Integer userId;
    private Boolean needFormat;
    private Integer[] suiteIds;
    private List<Integer> caseIds;
    private List<Integer> spaceIds;
    private String name;
    private String type;

    public ExportCaseTask(Integer[] suiteIds, List<Integer> caseIds, List<Integer> spaceIds, String name, List<Integer> priority, String type,boolean needFormat, Integer userId) {
        this.needFormat = needFormat;
        this.priority = priority;
        this.userId = userId;
        this.suiteIds = suiteIds;
        this.caseIds = caseIds;
        this.spaceIds = spaceIds;
        this.name = name;
        this.type = type;
    }

    private TestCaseService testCaseService = (TestCaseService) SpringContextUtils.getBean("testCaseService");

    public Object call() throws Exception {

        try {
            testCaseService.generateASync(suiteIds, caseIds, spaceIds, name, priority, type, needFormat, userId);
            logger.info("generateASync ok");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
        return 1;
    }
}