Java多任务多线程，总线程数countDownLatch限制模板（附源码）

问题背景
项目创建
总结
Lyric：在月光下一直找寻

问题背景

最近开发的项目需要多任务并行运行，然后每个任务需要多线程运行，要求如下：

多任务并行，一个任务可设置线程数
限制整个项目开启任务的线程数500，大于500则等待线程执行完毕，再进行创建线程
大于500是等待线程结束，最多等待10分钟
注意事项：
可以通过复制文章的代码自己创建工程，也可以下载源码进行参考

项目创建

1 引入pom依赖



    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        2.6.7
         
    
    com.yg
    taskFrame
    0.0.1-SNAPSHOT
    taskFrame
    Demo project for Spring Boot
    
        1.8
    
    
        
            org.springframework.boot
            spring-boot-starter-web
        

        
            org.projectlombok
            lombok
            true
        
        
            org.springframework.boot
            spring-boot-starter-test
            test
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
                
                    
                        
                            org.projectlombok
                            lombok

2 启动类

package com.yg.taskframe;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TaskFrameApplication {

    public static void main(String[] args) {
        SpringApplication.run(TaskFrameApplication.class, args);
    }

}

3 Runnable线程类

package com.yg.taskframe.core;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;

@Slf4j
public class CreateRunnable implements Runnable {
    private CountDownLatch countDownLatch;

    public CreateRunnable(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.error("CreateRunnable error: ", e);
        }
        countDownLatch.countDown();
    }
}

4 线程池管理服务类

package com.yg.taskframe.core;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.CountDownLatch;


@Slf4j
public class AsynThreadService {

    public ThreadPoolTaskExecutor asyncServiceExecutor;
    public CountDownLatch countDownLatch;

    public AsynThreadService(ThreadPoolTaskExecutor asyncServiceExecutor, CountDownLatch countDownLatch) {
        this.asyncServiceExecutor = asyncServiceExecutor;
        this.countDownLatch = countDownLatch;
    }


//    public void submit(ComputeDTO computeDTO, String pathCode, List param) {
//        asyncServiceExecutor.submit(new CreateRunnable(countDownLatch));
//    }

    public void waitComplete() {
        try {
            //等待当前线程池对象的所有countDownLatch都已经释放，说明线程都执行完毕了，然后关闭线程池
            log.info("countDownLatch remain {}", this.countDownLatch.getCount());
            this.countDownLatch.await();
            log.info("Close asynThreadService");
            asyncServiceExecutor.shutdown();
        } catch (Exception e) {
            log.error("AsyncServiceExecutor shutdown error: ", e);
        }
    }
}

5 线程池和 countDownLatch 创建类

package com.yg.taskframe.core;


import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


@Slf4j
public class AsynThreadPool {
    private static AtomicInteger curCount = new AtomicInteger(0);
    private static Integer maxCount = 500;

    public static AsynThreadService createCountDownLatch(Long taskId, int threads) throws InterruptedException {
        CountDownLatch = new CountDownLatch(threads);

        //创建新的线程池
        ThreadPoolTaskExecutor asyncServiceExecutor = creatThreadPool(taskId, threads);

        //超过了CountDownLatch总数，每次等待1一分钟，总共等待十次10分钟
        if (curCount.get() >= maxCount) {
            int n = 10;
            while (n-- > 0) {
                if (curCount.get() >= maxCount) {
                    log.info("CountDownLatch count more than 500, loading over for other threads");
                    //每一分钟检查一次，大于总的线程数，进行一分钟等待，然后继续查询是否大于总的线程数，一共等待10次
                    TimeUnit.SECONDS.sleep(60);
                } else {
                    //一旦发现 curCount.get() < maxCount，立马结束循环
                    n = 0;
                }
                if (n == 0) {
                    log.warn("Part of asynThreadService aren't finish");
                }
            }
        }
        //原子获取累加
        curCount.addAndGet(threads);

        AsynThreadService asynThreadService = new AsynThreadService(asyncServiceExecutor, countDownLatch);
        return asynThreadService;
    }

    //释放countDownLatch
    public static void free(int count) {
        log.info("CountDownLatch : {}, count: {}", curCount.get(), count);
        curCount.addAndGet(-count);
        log.info("CountDownLatch remain: {}", curCount.get());
    }


    //创建线程池
    public static ThreadPoolTaskExecutor creatThreadPool(Long taskId, int threads) {
        ThreadPoolTaskExecutor asyncServiceExecutor = null;
        try {
            asyncServiceExecutor = new ThreadPoolTaskExecutor();
            // 线程池维护线程的最少数量
            // asyncServiceExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() + 1);
            asyncServiceExecutor.setCorePoolSize(threads);
            // 线程池维护线程的最大数量
            asyncServiceExecutor.setMaxPoolSize(threads + 1);
            // 线程池所使用的缓冲队列
            asyncServiceExecutor.setQueueCapacity(2000);
            //   asyncServiceExecutor.prefersShortLivedTasks();
            asyncServiceExecutor.setThreadNamePrefix("TaskId" + taskId.toString() + "-Thread-");
            asyncServiceExecutor.setBeanName("TaskId" + taskId);
            //  asyncServiceExecutor.setKeepAliveSeconds(20);
            //调用者执行
            //   asyncServiceExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            asyncServiceExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
            // 线程全部结束才关闭线程池
            asyncServiceExecutor.setWaitForTasksToCompleteOnShutdown(true);
            // 如果超过60s还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住
            asyncServiceExecutor.setAwaitTerminationSeconds(30);
            asyncServiceExecutor.initialize();
        } catch (Exception e) {
            log.error("Create ThreadPoolTaskExecutor failed", e);
        }
        return asyncServiceExecutor;
    }
}

6 任务创建类

package com.yg.taskframe.core;

import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;


@Slf4j
@AllArgsConstructor
@NoArgsConstructor
public class ParallelTask {
    private int threads;
    private String url;

    public ParallelTask nextTask;
    public ParallelTask prevTask;

    public ParallelTask(int threads) {
        this.threads = threads;
    }

    public AsynThreadService startTask(Long taskId) throws InterruptedException {

        //任务里面线程大于1，就创建线程池，创建线程池要先判断所有线程池的总数是否大于500,
        AsynThreadService asynThreadService = AsynThreadPool.createCountDownLatch(taskId, threads);
        return asynThreadService;

    }
}

7 启动一个任务入口类

package com.yg.taskframe.service;

import com.yg.taskframe.core.AsynThreadPool;
import com.yg.taskframe.core.AsynThreadService;
import com.yg.taskframe.core.ParallelTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;



@Slf4j
@Service
public class TaskService {


    
    public void startTask(Long taskId, int threads) throws InterruptedException {

        log.info("Begin task");

        //可以通过配置设置线程数
       // int threads = threads;
        if (threads == 0) {
            threads = 1;
        }
        log.info("threads: {}", threads);
//        Random random = new Random();
//        //任务id可以通过配置下发
//        Long taskId = random.nextInt(100) + 500L;
        //初始化一个任务的线程数
        ParallelTask parallelTask = new ParallelTask(threads);
        //创建单个任务线程为threads的线程池，并设置任务名为taskId组合
        AsynThreadService asynThreadService = parallelTask.startTask(taskId);
        //单线程查询doquery
        List params = new ArrayList<>();
        params.add("1");
        submit(params, asynThreadService);

        //线程执行完成，关闭线程池，减去countDownLatch
        int countDownLatchCount = (int) asynThreadService.countDownLatch.getCount();
        while (countDownLatchCount-- > 0) {
            asynThreadService.countDownLatch.countDown();
        }
        asynThreadService.waitComplete();
        AsynThreadPool.free(threads);
    }


    public void submit(List params, AsynThreadService asynThreadService) {

        //执行线程
        log.info("Begin 0004 thread");
        Future futureResult = asynThreadService.asyncServiceExecutor.submit(() ->
                submitSingle(params, asynThreadService.countDownLatch)
        );

    }


    
    public String submitSingle(List params, CountDownLatch countDownLatch) {
        log.info("Submit single thread");
        try {
            log.info("params size: {}", params.size());
            long start = System.currentTimeMillis();
            //可以远程调用其他服务
            String resultVOlist = doQuery(params);
            log.info("resultVOlist:{}", resultVOlist);
            return resultVOlist;
        } catch (Exception e) {
            log.error("dataxQueryController not found error", e);
            return "dataxQueryController not found error";
        } finally {
            log.info("Close one countDownLatch");
            countDownLatch.countDown();
        }

    }


    public String doQuery(List params) {
        if (params.size() < 10) {
            return "success";
        }
        return "fail";
    }

}

8 使用接口添加任务，通过需要设置每个线程池的线程数和任务ID

package com.yg.taskframe.controller;

import com.yg.taskframe.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;



@Slf4j
@RestController
public class TaskController {

    @Autowired
    TaskService taskService;


    @PostMapping("/task")
    public String addTask(@RequestParam Long taskId, @RequestParam int threads) {
        try {
            taskService.startTask(taskId, threads);
            return "success";
        } catch (Exception e) {
            log.error("Exception", e);
            return "fail";
        }
    }
}

9 整个项目目录

总结

通过模板可以进行多任务并行运行，并且可以控制总线程数

作为程序员第 122 篇文章，每次写一句歌词记录一下，看看人生有几首歌的时间，wahahaha …

Lyric：在月光下一直找寻

Java多任务多线程，总线程数countDownLatch限制模板（附源码）

Java相关栏目本月热门文章